"""
Flask application exposing search, graph, and transcript endpoints for TLC.
Routes:
GET / -> static HTML search page.
GET /graph -> static reference graph UI.
GET /api/channels -> channels aggregation.
GET /api/channel-list -> canonical channel list + feed URL.
GET /channels.txt -> raw channel URLs list.
GET /api/search -> Elasticsearch keyword search.
GET /api/graph -> reference graph API.
GET /api/transcript -> transcript JSON payload.
"""
from __future__ import annotations
import copy
import json
import logging
import os
import re
import urllib.parse
import xml.etree.ElementTree as ET
from pathlib import Path
from typing import Any, Deque, Dict, Iterable, List, Optional, Sequence, Set, Tuple
from collections import Counter, defaultdict, deque
from datetime import datetime
from threading import Lock
from time import monotonic
import requests
from flask import Flask, jsonify, request, send_from_directory
from .config import CONFIG, AppConfig
from .channel_config import load_channel_entries
try:
from elasticsearch import Elasticsearch # type: ignore
from elasticsearch import BadRequestError # type: ignore
except ImportError: # pragma: no cover - dependency optional
Elasticsearch = None
BadRequestError = Exception # type: ignore
LOGGER = logging.getLogger(__name__)
# Security constants
MAX_QUERY_SIZE = 100
MAX_OFFSET = 10000
DEFAULT_ELASTIC_TIMEOUT = int(os.environ.get("ELASTIC_TIMEOUT_SECONDS", "30"))
_RATE_LIMIT_BUCKETS: Dict[str, Deque[float]] = defaultdict(deque)
_RATE_LIMIT_LOCK = Lock()
_RSS_AUTHOR_CACHE: Dict[str, Tuple[str, float]] = {}
_RSS_AUTHOR_LOCK = Lock()
_RSS_AUTHOR_TTL_SECONDS = 60 * 60 * 24
_RSS_OEMBED_LIMIT = 12
def _client_rate_key() -> str:
forwarded = request.headers.get("X-Forwarded-For", "")
if forwarded:
return forwarded.split(",")[0].strip()
return request.headers.get("X-Real-IP") or request.remote_addr or "unknown"
def _rate_limited_response(retry_after: int):
response = jsonify({"error": "rate_limited", "retryAfter": retry_after})
response.status_code = 429
response.headers["Retry-After"] = str(retry_after)
return response
def sanitize_query_string(query: str) -> str:
"""
Sanitize user input for Elasticsearch query_string queries.
Removes dangerous field targeting and script injection patterns.
"""
if not query:
return "*"
sanitized = query.strip()
# Remove field targeting patterns like "_id:", "_source:", "script:"
dangerous_field_patterns = [
r'\b_[a-z_]+\s*:', # Internal fields like _id:, _source:
r'\bscript\s*:', # Script injection
]
for pattern in dangerous_field_patterns:
sanitized = re.sub(pattern, '', sanitized, flags=re.IGNORECASE)
# Remove excessive wildcards that could cause ReDoS
sanitized = re.sub(r'\*{2,}', '*', sanitized)
sanitized = re.sub(r'\?{2,}', '?', sanitized)
return sanitized.strip() or "*"
def _ensure_client(config: AppConfig) -> "Elasticsearch":
if Elasticsearch is None:
raise RuntimeError(
"elasticsearch package not installed. "
"Install elasticsearch>=7 to run the Flask search app."
)
kwargs = {}
if config.elastic.api_key:
kwargs["api_key"] = config.elastic.api_key
elif config.elastic.username and config.elastic.password:
kwargs["basic_auth"] = (
config.elastic.username,
config.elastic.password,
)
if config.elastic.ca_cert:
kwargs["ca_certs"] = str(config.elastic.ca_cert)
kwargs["verify_certs"] = config.elastic.verify_certs
return Elasticsearch(config.elastic.url, **kwargs)
def _extract_video_id(url: str) -> Optional[str]:
if not url:
return None
try:
parsed = urllib.parse.urlparse(url.strip())
except Exception:
return None
host = (parsed.netloc or "").lower()
path = parsed.path or ""
if host in {"youtu.be", "www.youtu.be"}:
return path.lstrip("/") or None
if host.endswith("youtube.com"):
if path == "/watch":
params = urllib.parse.parse_qs(parsed.query)
return (params.get("v") or [None])[0]
if path.startswith("/shorts/"):
return path.split("/", 2)[2] if len(path.split("/", 2)) > 2 else None
return None
def _lookup_channel_names(
client: "Elasticsearch",
index: str,
video_ids: Iterable[str],
) -> Dict[str, str]:
ids = [vid for vid in video_ids if vid]
if not ids:
return {}
now = monotonic()
mapping: Dict[str, str] = {}
cached_hits = 0
elastic_hits = 0
remaining = []
with _RSS_AUTHOR_LOCK:
for vid in ids:
cached = _RSS_AUTHOR_CACHE.get(vid)
if cached and (now - cached[1]) < _RSS_AUTHOR_TTL_SECONDS:
mapping[vid] = cached[0]
cached_hits += 1
else:
remaining.append(vid)
if remaining:
try:
response = client.mget(index=index, body={"ids": remaining})
except Exception as exc: # pragma: no cover - elasticsearch handles errors
LOGGER.warning("RSS title lookup failed: %s", exc)
response = {}
for doc in response.get("docs", []):
if not doc.get("found"):
continue
source = doc.get("_source") or {}
name = source.get("channel_name") or source.get("channel_id")
if name:
vid = doc.get("_id", "")
mapping[vid] = str(name)
elastic_hits += 1
with _RSS_AUTHOR_LOCK:
_RSS_AUTHOR_CACHE[vid] = (mapping[vid], now)
missing = [vid for vid in remaining if vid not in mapping]
oembed_hits = 0
oembed_attempts = 0
if missing:
for vid in missing[:_RSS_OEMBED_LIMIT]:
oembed_attempts += 1
video_url = f"https://www.youtube.com/watch?v={vid}"
oembed_url = (
"https://www.youtube.com/oembed?format=json&url="
+ urllib.parse.quote(video_url, safe="")
)
try:
response = requests.get(oembed_url, timeout=10)
if response.status_code != 200:
continue
data = response.json()
except Exception:
continue
author = data.get("author_name")
if not author:
continue
mapping[vid] = str(author)
oembed_hits += 1
with _RSS_AUTHOR_LOCK:
_RSS_AUTHOR_CACHE[vid] = (mapping[vid], now)
missing_count = max(len(ids) - cached_hits - elastic_hits - oembed_hits, 0)
if missing_count or oembed_attempts:
LOGGER.info(
"RSS title lookup: total=%d cached=%d elastic=%d oembed=%d missing=%d",
len(ids),
cached_hits,
elastic_hits,
oembed_hits,
missing_count,
)
else:
LOGGER.debug(
"RSS title lookup: total=%d cached=%d elastic=%d",
len(ids),
cached_hits,
elastic_hits,
)
return mapping
def _rewrite_rss_payload(
content: bytes,
client: "Elasticsearch",
index: str,
feed_name: str,
) -> bytes:
try:
root = ET.fromstring(content)
except ET.ParseError:
LOGGER.warning("RSS rewrite skipped (invalid XML) for %s", feed_name)
return content
channel = root.find("channel")
if channel is None:
LOGGER.warning("RSS rewrite skipped (missing channel) for %s", feed_name)
return content
items = channel.findall("item")
total_items = len(items)
removed_errors = 0
video_ids: Set[str] = set()
for item in list(items):
title_el = item.find("title")
title_text = (title_el.text or "").strip() if title_el is not None else ""
if "Bridge returned error" in title_text:
channel.remove(item)
removed_errors += 1
continue
link_el = item.find("link")
guid_el = item.find("guid")
video_id = _extract_video_id((link_el.text or "") if link_el is not None else "")
if not video_id:
video_id = _extract_video_id((guid_el.text or "") if guid_el is not None else "")
if video_id:
video_ids.add(video_id)
channel_name_map = _lookup_channel_names(client, index, video_ids)
if not channel_name_map:
LOGGER.info(
"RSS rewrite: feed=%s items=%d removed_errors=%d resolved=0",
feed_name,
total_items,
removed_errors,
)
return ET.tostring(root, encoding="utf-8", xml_declaration=True)
prefixed = 0
for item in channel.findall("item"):
title_el = item.find("title")
if title_el is None or not title_el.text:
continue
link_el = item.find("link")
guid_el = item.find("guid")
video_id = _extract_video_id((link_el.text or "") if link_el is not None else "")
if not video_id:
video_id = _extract_video_id((guid_el.text or "") if guid_el is not None else "")
if not video_id:
continue
channel_name = channel_name_map.get(video_id)
if not channel_name:
continue
prefix = f"{channel_name} - "
if title_el.text.startswith(prefix):
continue
title_el.text = f"{channel_name} - {title_el.text}"
prefixed += 1
LOGGER.info(
"RSS rewrite: feed=%s items=%d removed_errors=%d prefixed=%d resolved=%d",
feed_name,
total_items,
removed_errors,
prefixed,
len(channel_name_map),
)
return ET.tostring(root, encoding="utf-8", xml_declaration=True)
def metrics_payload(data_root: Path, include_external: bool = True) -> Dict[str, Any]:
total_items = 0
channel_counter: Counter = Counter()
channel_name_map: Dict[str, str] = {}
year_counter: Counter = Counter()
month_counter: Counter = Counter()
if not data_root.exists():
LOGGER.warning("Data directory %s not found; metrics will be empty.", data_root)
return {
"totalItems": 0,
"totalChannels": 0,
"itemsPerChannel": [],
"yearHistogram": [],
"recentMonths": [],
}
for path in data_root.rglob("*.json"):
try:
with path.open("r", encoding="utf-8") as handle:
doc = json.load(handle)
except Exception:
continue
if not include_external and doc.get("external_reference"):
continue
total_items += 1
channel_id = doc.get("channel_id")
channel_name = doc.get("channel_name") or channel_id
if channel_id:
channel_counter[channel_id] += 1
if channel_name and channel_id not in channel_name_map:
channel_name_map[channel_id] = channel_name
date_value = doc.get("date") or doc.get("published_at")
dt: Optional[datetime] = None
if isinstance(date_value, str):
for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"):
try:
dt = datetime.strptime(date_value[: len(fmt)], fmt)
break
except Exception:
continue
elif isinstance(date_value, (int, float)):
try:
dt = datetime.fromtimestamp(date_value)
except Exception:
dt = None
if dt:
year_counter[str(dt.year)] += 1
month_counter[dt.strftime("%Y-%m")] += 1
items_per_channel = [
{
"label": channel_name_map.get(cid, cid),
"count": count,
}
for cid, count in channel_counter.most_common()
]
year_histogram = [
{"bucket": year, "count": year_counter[year]}
for year in sorted(year_counter.keys())
]
recent_months = sorted(month_counter.keys())
recent_months = recent_months[-12:]
recent_months_payload = [
{"bucket": month, "count": month_counter[month]} for month in recent_months
]
return {
"totalItems": total_items,
"totalChannels": len(channel_counter),
"itemsPerChannel": items_per_channel,
"yearHistogram": year_histogram,
"recentMonths": recent_months_payload,
}
def elastic_metrics_payload(
client: "Elasticsearch",
index: str,
*,
channel_field_candidates: Optional[List[str]] = None,
debug: bool = False,
include_external: bool = True,
) -> Dict[str, Any]:
if channel_field_candidates is None:
channel_field_candidates = ["channel_id.keyword", "channel_id"]
base_body: Dict[str, Any] = {
"size": 0,
"track_total_hits": True,
"aggs": {
"channels": {
"terms": {
"field": "channel_id.keyword",
"size": 500,
"order": {"_count": "desc"},
},
"aggs": {
"name": {
"top_hits": {
"size": 1,
"_source": {"includes": ["channel_name"]},
"sort": [
{
"channel_name.keyword": {
"order": "asc",
"missing": "_last",
"unmapped_type": "keyword",
}
}
],
}
}
},
},
"year_histogram": {
"date_histogram": {
"field": "date",
"calendar_interval": "year",
"format": "yyyy",
}
},
"month_histogram": {
"date_histogram": {
"field": "date",
"calendar_interval": "month",
"format": "yyyy-MM",
"order": {"_key": "asc"},
}
},
},
}
if not include_external:
base_body["query"] = {"bool": {"must_not": [{"term": {"external_reference": True}}]}}
last_error: Optional[Exception] = None
response: Optional[Dict[str, Any]] = None
for candidate_field in channel_field_candidates:
body = json.loads(json.dumps(base_body))
body["aggs"]["channels"]["terms"]["field"] = candidate_field
try:
if debug:
LOGGER.info(
"Elasticsearch metrics request: %s",
json.dumps({"index": index, "body": body}, indent=2),
)
response = client.search(index=index, body=body, request_timeout=30)
break
except BadRequestError as exc:
last_error = exc
if debug:
LOGGER.warning(
"Metrics aggregation failed for field %s: %s",
candidate_field,
exc,
)
if response is None:
raise last_error or RuntimeError("Unable to compute metrics from Elasticsearch.")
hits = response.get("hits", {})
total_items = hits.get("total", {}).get("value", 0)
if debug:
LOGGER.info(
"Elasticsearch metrics response: %s",
json.dumps(response, indent=2, default=str),
)
aggregations = response.get("aggregations", {})
channel_buckets = aggregations.get("channels", {}).get("buckets", [])
items_per_channel = []
for bucket in channel_buckets:
key = bucket.get("key")
channel_name = key
top_hits = (
bucket.get("name", {})
.get("hits", {})
.get("hits", [])
)
if top_hits:
channel_name = (
top_hits[0]
.get("_source", {})
.get("channel_name", channel_name)
)
items_per_channel.append(
{"label": channel_name or key, "count": bucket.get("doc_count", 0)}
)
year_buckets = aggregations.get("year_histogram", {}).get("buckets", [])
year_histogram = [
{
"bucket": bucket.get("key_as_string")
or str(bucket.get("key")),
"count": bucket.get("doc_count", 0),
}
for bucket in year_buckets
]
month_buckets = aggregations.get("month_histogram", {}).get("buckets", [])
recent_months_entries = [
{
"bucket": bucket.get("key_as_string")
or str(bucket.get("key")),
"count": bucket.get("doc_count", 0),
"_key": bucket.get("key"),
}
for bucket in month_buckets
]
recent_months_entries.sort(key=lambda item: item.get("_key", 0))
recent_months_payload = [
{"bucket": entry["bucket"], "count": entry["count"]}
for entry in recent_months_entries[-12:]
]
return {
"totalItems": total_items,
"totalChannels": len(items_per_channel),
"itemsPerChannel": items_per_channel,
"yearHistogram": year_histogram,
"recentMonths": recent_months_payload,
}
def parse_channel_params(values: Iterable[Optional[str]]) -> List[str]:
seen: Set[str] = set()
channels: List[str] = []
for value in values:
if not value:
continue
for part in str(value).split(","):
cleaned = part.strip()
if not cleaned or cleaned.lower() == "all":
continue
if cleaned not in seen:
seen.add(cleaned)
channels.append(cleaned)
return channels
def build_year_filter(year: Optional[str]) -> Optional[Dict]:
if not year:
return None
try:
year_int = int(year)
return {
"range": {
"date": {
"gte": f"{year_int}-01-01",
"lt": f"{year_int + 1}-01-01",
"format": "yyyy-MM-dd"
}
}
}
except (ValueError, TypeError):
return None
def build_channel_filter(channels: Optional[Sequence[str]]) -> Optional[Dict]:
if not channels:
return None
per_channel_clauses: List[Dict[str, Any]] = []
for value in channels:
if not value:
continue
per_channel_clauses.append(
{
"bool": {
"should": [
{"term": {"channel_id.keyword": value}},
{"term": {"channel_id": value}},
],
"minimum_should_match": 1,
}
}
)
if not per_channel_clauses:
return None
if len(per_channel_clauses) == 1:
return per_channel_clauses[0]
return {
"bool": {
"should": per_channel_clauses,
"minimum_should_match": 1,
}
}
def build_query_payload(
query: str,
*,
channels: Optional[Sequence[str]] = None,
year: Optional[str] = None,
sort: str = "relevant",
use_exact: bool = True,
use_fuzzy: bool = True,
use_phrase: bool = True,
use_query_string: bool = False,
include_external: bool = True,
) -> Dict:
filters: List[Dict] = []
should: List[Dict] = []
channel_filter = build_channel_filter(channels)
if channel_filter:
filters.append(channel_filter)
year_filter = build_year_filter(year)
if year_filter:
filters.append(year_filter)
if not include_external:
filters.append({"bool": {"must_not": [{"term": {"external_reference": True}}]}})
if use_query_string:
base_fields = ["title^3", "description^2", "transcript_full", "transcript_secondary_full"]
qs_query = sanitize_query_string(query or "")
query_body: Dict[str, Any] = {
"query_string": {
"query": qs_query,
"default_operator": "AND",
"fields": base_fields,
}
}
if filters:
query_body = {"bool": {"must": query_body, "filter": filters}}
body: Dict = {
"query": query_body,
"highlight": {
"fields": {
"transcript_full": {
"fragment_size": 160,
"number_of_fragments": 5,
"fragmenter": "span",
},
"transcript_secondary_full": {
"fragment_size": 160,
"number_of_fragments": 5,
"fragmenter": "span",
},
"title": {"number_of_fragments": 0},
"description": {
"fragment_size": 160,
"number_of_fragments": 1,
},
},
"require_field_match": False,
"pre_tags": [""],
"post_tags": [""],
"encoder": "html",
"max_analyzed_offset": 900000,
},
}
if sort == "newer":
body["sort"] = [{"date": {"order": "desc"}}]
elif sort == "older":
body["sort"] = [{"date": {"order": "asc"}}]
elif sort == "referenced":
body["sort"] = [{"referenced_by_count": {"order": "desc", "unmapped_type": "long"}}]
return body
if query:
base_fields = ["title^3", "description^2", "transcript_full", "transcript_secondary_full"]
if use_phrase:
should.append(
{
"match_phrase": {
"transcript_full": {
"query": query,
"slop": 2,
"boost": 10.0,
}
}
}
)
should.append(
{
"match_phrase": {
"transcript_secondary_full": {
"query": query,
"slop": 2,
"boost": 10.0,
}
}
}
)
should.append(
{
"match_phrase": {
"title": {
"query": query,
"slop": 0,
"boost": 50.0,
}
}
}
)
if use_fuzzy:
should.append(
{
"multi_match": {
"query": query,
"fields": base_fields,
"type": "best_fields",
"operator": "and",
"fuzziness": "AUTO",
"prefix_length": 1,
"max_expansions": 50,
"boost": 1.5,
}
}
)
if use_exact:
should.append(
{
"multi_match": {
"query": query,
"fields": base_fields,
"type": "best_fields",
"operator": "and",
"boost": 3.0,
}
}
)
if should:
query_body: Dict = {
"bool": {
"should": should,
"minimum_should_match": 1,
}
}
if filters:
query_body["bool"]["filter"] = filters
elif filters:
query_body = {"bool": {"filter": filters}}
else:
query_body = {"match_all": {}}
body: Dict = {
"query": query_body,
"highlight": {
"fields": {
"transcript_full": {
"fragment_size": 160,
"number_of_fragments": 5,
"fragmenter": "span",
},
"transcript_secondary_full": {
"fragment_size": 160,
"number_of_fragments": 5,
"fragmenter": "span",
},
"title": {"number_of_fragments": 0},
"description": {
"fragment_size": 160,
"number_of_fragments": 1,
},
},
"require_field_match": False,
"pre_tags": [""],
"post_tags": [""],
"encoder": "html",
"max_analyzed_offset": 900000,
},
}
if query_body.get("match_all") is None:
body["highlight"]["highlight_query"] = copy.deepcopy(query_body)
if sort == "newer":
body["sort"] = [{"date": {"order": "desc"}}]
elif sort == "older":
body["sort"] = [{"date": {"order": "asc"}}]
elif sort == "referenced":
body["sort"] = [{"referenced_by_count": {"order": "desc", "unmapped_type": "long"}}]
return body
def build_graph_payload(
client: "Elasticsearch",
index: str,
root_id: str,
depth: int,
max_nodes: int,
*,
include_external: bool = True,
) -> Dict[str, Any]:
root_id = root_id.strip()
if not root_id:
return {"nodes": [], "links": [], "root": root_id, "depth": depth, "meta": {}}
doc_cache: Dict[str, Optional[Dict[str, Any]]] = {}
def fetch_document(video_id: str) -> Optional[Dict[str, Any]]:
if video_id in doc_cache:
return doc_cache[video_id]
try:
result = client.get(index=index, id=video_id)
doc_cache[video_id] = result.get("_source")
except Exception as exc: # pragma: no cover - elasticsearch handles errors
LOGGER.debug("Graph: failed to load %s: %s", video_id, exc)
doc_cache[video_id] = None
doc = doc_cache[video_id]
if doc is not None and not include_external and doc.get("external_reference"):
doc_cache[video_id] = None
return None
return doc_cache[video_id]
nodes: Dict[str, Dict[str, Any]] = {}
links: List[Dict[str, Any]] = []
link_seen: Set[Tuple[str, str, str]] = set()
queue: deque[Tuple[str, int]] = deque([(root_id, 0)])
queued: Set[str] = {root_id}
visited: Set[str] = set()
while queue and len(nodes) < max_nodes:
current_id, level = queue.popleft()
queued.discard(current_id)
if current_id in visited:
continue
doc = fetch_document(current_id)
if doc is None:
if current_id == root_id:
break
visited.add(current_id)
continue
visited.add(current_id)
nodes[current_id] = {
"id": current_id,
"title": doc.get("title") or current_id,
"channel_id": doc.get("channel_id"),
"channel_name": doc.get("channel_name") or doc.get("channel_id") or "Unknown",
"url": doc.get("url"),
"date": doc.get("date"),
"is_root": current_id == root_id,
"external_reference": bool(doc.get("external_reference")),
}
if level >= depth:
continue
neighbor_ids: List[str] = []
for ref_id in normalize_reference_list(doc.get("internal_references")):
if ref_id == current_id:
continue
key = (current_id, ref_id, "references")
if key not in link_seen:
links.append(
{"source": current_id, "target": ref_id, "relation": "references"}
)
link_seen.add(key)
neighbor_ids.append(ref_id)
for ref_id in normalize_reference_list(doc.get("referenced_by")):
if ref_id == current_id:
continue
key = (ref_id, current_id, "referenced_by")
if key not in link_seen:
links.append(
{"source": ref_id, "target": current_id, "relation": "referenced_by"}
)
link_seen.add(key)
neighbor_ids.append(ref_id)
for neighbor in neighbor_ids:
if neighbor in visited or neighbor in queued:
continue
if len(nodes) + len(queue) >= max_nodes:
break
queue.append((neighbor, level + 1))
queued.add(neighbor)
# Ensure nodes referenced by links exist in the payload.
for link in links:
for key in ("source", "target"):
node_id = link[key]
if node_id in nodes:
continue
doc = fetch_document(node_id)
if doc is None:
if include_external:
nodes[node_id] = {
"id": node_id,
"title": node_id,
"channel_id": None,
"channel_name": "Unknown",
"url": None,
"date": None,
"is_root": node_id == root_id,
"external_reference": False,
}
continue
nodes[node_id] = {
"id": node_id,
"title": doc.get("title") or node_id,
"channel_id": doc.get("channel_id"),
"channel_name": doc.get("channel_name") or doc.get("channel_id") or "Unknown",
"url": doc.get("url"),
"date": doc.get("date"),
"is_root": node_id == root_id,
"external_reference": bool(doc.get("external_reference")),
}
links = [
link
for link in links
if link.get("source") in nodes and link.get("target") in nodes
]
return {
"root": root_id,
"depth": depth,
"nodes": list(nodes.values()),
"links": links,
"meta": {
"node_count": len(nodes),
"link_count": len(links),
},
}
def build_full_graph_payload(
client: "Elasticsearch",
index: str,
max_nodes: Optional[int],
*,
highlight_id: Optional[str] = None,
include_external: bool = True,
) -> Dict[str, Any]:
"""
Attempt to render the entire reference graph by gathering every video that
references another (or is referenced).
"""
query = {
"bool": {
"should": [
{"range": {"internal_references_count": {"gt": 0}}},
{"range": {"referenced_by_count": {"gt": 0}}},
{"exists": {"field": "internal_references"}},
{"exists": {"field": "referenced_by"}},
],
"minimum_should_match": 1,
}
}
source_fields = [
"video_id",
"title",
"channel_id",
"channel_name",
"url",
"date",
"internal_references",
"referenced_by",
]
nodes: Dict[str, Dict[str, Any]] = {}
links: List[Dict[str, Any]] = []
link_seen: Set[Tuple[str, str, str]] = set()
node_limit = max_nodes if max_nodes and max_nodes > 0 else None
batch_size = 500 if node_limit is None else min(500, max(50, node_limit * 2))
truncated = False
def ensure_node(node_id: Optional[str], doc: Optional[Dict[str, Any]] = None) -> bool:
if not node_id:
return False
if node_id in nodes:
if doc:
if not include_external and doc.get("external_reference"):
nodes.pop(node_id, None)
return False
existing = nodes[node_id]
if (not existing.get("title") or existing["title"] == node_id) and doc.get("title"):
existing["title"] = doc["title"]
if not existing.get("channel_id") and doc.get("channel_id"):
existing["channel_id"] = doc.get("channel_id")
if (
existing.get("channel_name") in {"Unknown", node_id, None}
and (doc.get("channel_name") or doc.get("channel_id"))
):
existing["channel_name"] = doc.get("channel_name") or doc.get("channel_id")
if not existing.get("url") and doc.get("url"):
existing["url"] = doc.get("url")
if not existing.get("date") and doc.get("date"):
existing["date"] = doc.get("date")
return True
if node_limit is not None and len(nodes) >= node_limit:
return False
channel_name = None
channel_id = None
url = None
date_val = None
title = node_id
if doc:
if not include_external and doc.get("external_reference"):
return False
title = doc.get("title") or title
channel_id = doc.get("channel_id")
channel_name = doc.get("channel_name") or channel_id
url = doc.get("url")
date_val = doc.get("date")
nodes[node_id] = {
"id": node_id,
"title": title,
"channel_id": channel_id,
"channel_name": channel_name or "Unknown",
"url": url,
"date": date_val,
"is_root": False,
"external_reference": bool(doc.get("external_reference")) if doc else False,
}
return True
scroll_id: Optional[str] = None
try:
body = {"query": query, "_source": source_fields, "sort": ["_doc"]}
response = client.search(index=index, body=body, size=batch_size, scroll="1m", request_timeout=60)
scroll_id = response.get("_scroll_id")
stop_fetch = False
while not stop_fetch:
hits = response.get("hits", {}).get("hits", [])
if not hits:
break
for hit in hits:
if node_limit is not None and len(nodes) >= node_limit:
truncated = True
stop_fetch = True
break
source = hit.get("_source", {}) or {}
video_id = source.get("video_id")
if not video_id:
continue
if not include_external and source.get("external_reference"):
nodes.pop(video_id, None)
continue
if not ensure_node(video_id, source):
continue
for target in normalize_reference_list(source.get("internal_references")):
if target == video_id:
continue
if not ensure_node(target):
continue
key = (video_id, target, "references")
if key not in link_seen:
links.append(
{"source": video_id, "target": target, "relation": "references"}
)
link_seen.add(key)
for origin in normalize_reference_list(source.get("referenced_by")):
if origin == video_id:
continue
if not ensure_node(origin):
continue
key = (origin, video_id, "referenced_by")
if key not in link_seen:
links.append(
{"source": origin, "target": video_id, "relation": "referenced_by"}
)
link_seen.add(key)
if stop_fetch or not scroll_id:
break
response = client.scroll(scroll_id=scroll_id, scroll="1m")
scroll_id = response.get("_scroll_id")
if not scroll_id:
break
finally:
if scroll_id:
try:
client.clear_scroll(scroll_id=scroll_id)
except Exception:
pass
if highlight_id and highlight_id in nodes:
nodes[highlight_id]["is_root"] = True
links = [
link
for link in links
if link.get("source") in nodes and link.get("target") in nodes
]
return {
"root": highlight_id or "",
"depth": 0,
"nodes": list(nodes.values()),
"links": links,
"meta": {
"node_count": len(nodes),
"link_count": len(links),
"mode": "full",
"truncated": truncated,
},
}
def create_app(config: AppConfig = CONFIG) -> Flask:
app = Flask(__name__, static_folder=str(Path(__file__).parent / "static"))
app.config['MAX_CONTENT_LENGTH'] = 1 * 1024 * 1024
@app.after_request
def add_security_headers(response):
response.headers['X-Frame-Options'] = 'DENY'
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['Permissions-Policy'] = 'geolocation=(), microphone=(), camera=()'
response.headers['Content-Security-Policy'] = (
"default-src 'self'; "
"script-src 'self' https://cdn.jsdelivr.net https://unpkg.com; "
"style-src 'self' 'unsafe-inline' https://unpkg.com; "
"img-src 'self' data: https:; "
"font-src 'self' https://unpkg.com; "
"connect-src 'self'"
)
return response
@app.before_request
def enforce_rate_limit():
if not config.rate_limit.enabled:
return None
if not request.path.startswith("/api/"):
return None
limit = config.rate_limit.requests
window_seconds = config.rate_limit.window_seconds
if limit <= 0 or window_seconds <= 0:
return None
now = monotonic()
key = _client_rate_key()
with _RATE_LIMIT_LOCK:
bucket = _RATE_LIMIT_BUCKETS[key]
while bucket and now - bucket[0] > window_seconds:
bucket.popleft()
if len(bucket) >= limit:
retry_after = max(1, int(window_seconds - (now - bucket[0])))
return _rate_limited_response(retry_after)
bucket.append(now)
return None
client = _ensure_client(config)
index = config.elastic.index
@app.route("/")
def index_page():
return send_from_directory(app.static_folder, "index.html")
@app.route("/graph")
def graph_page():
return send_from_directory(app.static_folder, "graph.html")
@app.route("/notes")
def notes_page():
return send_from_directory(app.static_folder, "notes.html")
@app.route("/static/")
def static_files(filename: str):
return send_from_directory(app.static_folder, filename)
def normalize_reference_list(values: Any) -> List[str]:
if values is None:
return []
if isinstance(values, (list, tuple, set)):
iterable = values
else:
iterable = [values]
normalized: List[str] = []
for item in iterable:
candidate: Optional[str]
if isinstance(item, dict):
candidate = item.get("video_id") or item.get("id") # type: ignore[assignment]
else:
candidate = item # type: ignore[assignment]
if candidate is None:
continue
text = str(candidate).strip()
if not text:
continue
if text.lower() in {"none", "null"}:
continue
normalized.append(text)
return normalized
@app.route("/api/channels")
def channels():
include_external = request.args.get("external", default="0", type=str)
include_external = include_external.lower() not in {"0", "false", "no"}
base_channels_body = {
"size": 0,
"aggs": {
"channels": {
"terms": {"field": "channel_id", "size": 200},
"aggs": {
"name": {
"top_hits": {
"size": 1,
"_source": {"includes": ["channel_name"]},
"sort": [
{
"channel_name.keyword": {
"order": "asc",
"missing": "_last",
"unmapped_type": "keyword",
}
}
],
}
}
},
}
},
}
if not include_external:
base_channels_body["query"] = {
"bool": {"must_not": [{"term": {"external_reference": True}}]}
}
def run_channels_request(field_name: str):
body = json.loads(json.dumps(base_channels_body)) # deep copy
body["aggs"]["channels"]["terms"]["field"] = field_name
if config.elastic.debug:
LOGGER.info(
"Elasticsearch channels request: %s",
json.dumps({"index": index, "body": body}, indent=2),
)
return client.search(index=index, body=body)
response = None
last_error = None
for candidate_field in ("channel_id.keyword", "channel_id"):
try:
response = run_channels_request(candidate_field)
if config.elastic.debug:
LOGGER.info("Channels aggregation used field: %s", candidate_field)
break
except BadRequestError as exc:
last_error = exc
if config.elastic.debug:
LOGGER.warning(
"Channels aggregation failed for field %s: %s",
candidate_field,
exc,
)
if response is None:
raise last_error or RuntimeError("Unable to aggregate channels.")
if config.elastic.debug:
LOGGER.info(
"Elasticsearch channels response: %s",
json.dumps(response, indent=2, default=str),
)
buckets = (
response.get("aggregations", {})
.get("channels", {})
.get("buckets", [])
)
data = []
for bucket in buckets:
key = bucket.get("key")
name_hit = (
bucket.get("name", {})
.get("hits", {})
.get("hits", [{}])[0]
.get("_source", {})
.get("channel_name")
)
display_name = name_hit or key or "Unknown"
data.append(
{
"Id": key,
"Name": display_name,
"Count": bucket.get("doc_count", 0),
}
)
data.sort(key=lambda item: item["Name"].lower())
return jsonify(data)
@app.route("/api/channel-list")
def channel_list():
payload = {
"channels": [],
"rss_feed_url": config.rss_feed_url,
"source": str(config.channels_path),
}
try:
payload["channels"] = load_channel_entries(config.channels_path)
except FileNotFoundError:
LOGGER.warning("Channel list not found: %s", config.channels_path)
payload["error"] = "channels_not_found"
except Exception as exc:
LOGGER.exception("Failed to load channel list: %s", exc)
payload["error"] = "channels_load_failed"
return jsonify(payload)
@app.route("/channels.txt")
def channel_urls():
try:
channels = load_channel_entries(config.channels_path)
except FileNotFoundError:
LOGGER.warning("Channel list not found: %s", config.channels_path)
return jsonify({"error": "channels_not_found"}), 404
except Exception as exc:
LOGGER.exception("Failed to load channel list: %s", exc)
return jsonify({"error": "channels_load_failed"}), 500
urls = [channel["url"] for channel in channels if channel.get("url")]
body = "\n".join(urls) + ("\n" if urls else "")
return (body, 200, {"Content-Type": "text/plain; charset=utf-8"})
@app.route("/channels")
def channels_page():
try:
entries = load_channel_entries(config.channels_path)
except FileNotFoundError:
return "Channel list not found.", 404
except Exception:
return "Failed to load channel list.", 500
rows = ""
for ch in entries:
name = ch.get("name") or ch.get("handle") or ch.get("id") or "Unknown"
url = ch.get("url", "")
# Link to the channel page, not the /videos tab
channel_url = url.replace("/videos", "") if url.endswith("/videos") else url
name_html = (
f'{name}'
if channel_url
else name
)
rows += f"| {name_html} |
\n"
html = f"""
Channels - This Little Corner
Channels
{len(entries)} channels tracked
"""
return (html, 200, {"Content-Type": "text/html; charset=utf-8"})
def _rss_target(feed_name: str) -> str:
name = (feed_name or "").strip("/")
if not name:
name = "youtube-unified"
return f"{config.rss_feed_upstream.rstrip('/')}/rss/{name}"
@app.route("/rss")
@app.route("/rss/")
def rss_proxy(feed_name: str = ""):
target = _rss_target(feed_name)
try:
upstream = requests.get(target, timeout=30)
except requests.RequestException as exc:
LOGGER.warning("RSS upstream error for %s: %s", target, exc)
return jsonify({"error": "rss_unavailable"}), 502
payload = _rewrite_rss_payload(upstream.content, client, index, feed_name or "rss")
headers = {
"Content-Type": upstream.headers.get(
"Content-Type", "application/xml; charset=UTF-8"
)
}
cache_header = upstream.headers.get("Cache-Control")
if cache_header:
headers["Cache-Control"] = cache_header
etag = upstream.headers.get("ETag")
if etag:
headers["ETag"] = etag
last_modified = upstream.headers.get("Last-Modified")
if last_modified:
headers["Last-Modified"] = last_modified
return (payload, upstream.status_code, headers)
@app.route("/api/graph")
def graph_api():
video_id = (request.args.get("video_id") or "").strip()
full_graph = request.args.get("full_graph", default="0", type=str)
full_graph = full_graph.lower() in {"1", "true", "yes"}
if not full_graph and not video_id:
return jsonify({"error": "video_id is required"}), 400
include_external = request.args.get("external", default="1", type=str)
include_external = include_external.lower() not in {"0", "false", "no"}
try:
depth = int(request.args.get("depth", "1"))
except ValueError:
depth = 1
depth = max(0, min(depth, 3))
if full_graph:
max_nodes = None
else:
try:
max_nodes = int(request.args.get("max_nodes", "200"))
except ValueError:
max_nodes = 200
max_nodes = max(10, min(max_nodes, 400))
if full_graph:
payload = build_full_graph_payload(
client,
index,
None,
highlight_id=video_id or None,
include_external=include_external,
)
else:
payload = build_graph_payload(
client,
index,
video_id,
depth,
max_nodes,
include_external=include_external,
)
if not payload["nodes"]:
return (
jsonify({"error": f"Video '{video_id}' was not found in the index."}),
404,
)
payload["meta"]["max_nodes"] = (
len(payload["nodes"]) if full_graph else max_nodes
)
return jsonify(payload)
@app.route("/api/years")
def years():
body = {
"size": 0,
"aggs": {
"years": {
"date_histogram": {
"field": "date",
"calendar_interval": "year",
"format": "yyyy",
"order": {"_key": "desc"}
}
}
}
}
if config.elastic.debug:
LOGGER.info(
"Elasticsearch years request: %s",
json.dumps({"index": index, "body": body}, indent=2),
)
response = client.search(index=index, body=body)
if config.elastic.debug:
LOGGER.info(
"Elasticsearch years response: %s",
json.dumps(response, indent=2, default=str),
)
buckets = (
response.get("aggregations", {})
.get("years", {})
.get("buckets", [])
)
data = [
{
"Year": bucket.get("key_as_string"),
"Count": bucket.get("doc_count", 0),
}
for bucket in buckets
if bucket.get("doc_count", 0) > 0
]
return jsonify(data)
@app.route("/api/search")
def search():
query = request.args.get("q", "", type=str)
raw_channels: List[Optional[str]] = request.args.getlist("channel_id")
legacy_channel = request.args.get("channel", type=str)
if legacy_channel:
raw_channels.append(legacy_channel)
channels = parse_channel_params(raw_channels)
year = request.args.get("year", "", type=str) or None
sort = request.args.get("sort", "relevant", type=str)
page = max(request.args.get("page", 0, type=int), 0)
size = min(max(request.args.get("size", 10, type=int), 1), MAX_QUERY_SIZE)
def parse_flag(name: str, default: bool = True) -> bool:
value = request.args.get(name)
if value is None:
return default
return value.lower() not in {"0", "false", "no"}
use_exact = parse_flag("exact", True)
use_fuzzy = parse_flag("fuzzy", True)
use_phrase = parse_flag("phrase", True)
use_query_string = parse_flag("query_string", False)
include_external = parse_flag("external", False)
if use_query_string:
use_exact = use_fuzzy = use_phrase = False
payload = build_query_payload(
query,
channels=channels,
year=year,
sort=sort,
use_exact=use_exact,
use_fuzzy=use_fuzzy,
use_phrase=use_phrase,
use_query_string=use_query_string,
include_external=include_external,
)
start = page * size
if start >= MAX_OFFSET:
return jsonify({"error": "offset_too_large", "maxOffset": MAX_OFFSET}), 400
if start + size > MAX_OFFSET:
size = max(1, MAX_OFFSET - start)
if config.elastic.debug:
LOGGER.info(
"Elasticsearch search request: %s",
json.dumps(
{
"index": index,
"from": start,
"size": size,
"body": payload,
"channels": channels,
"toggles": {
"exact": use_exact,
"fuzzy": use_fuzzy,
"phrase": use_phrase,
},
},
indent=2,
),
)
response = client.search(
index=index,
from_=start,
size=size,
body=payload,
request_timeout=30,
)
if config.elastic.debug:
LOGGER.info(
"Elasticsearch search response: %s",
json.dumps(response, indent=2, default=str),
)
hits = response.get("hits", {})
total = hits.get("total", {}).get("value", 0)
documents = []
for hit in hits.get("hits", []):
source = hit.get("_source", {})
highlight_map = hit.get("highlight", {})
transcript_highlight = [
{"html": value, "source": "primary"}
for value in (highlight_map.get("transcript_full", []) or [])
] + [
{"html": value, "source": "secondary"}
for value in (highlight_map.get("transcript_secondary_full", []) or [])
]
title_highlight = highlight_map.get("title") or []
description_highlight = highlight_map.get("description") or []
title_html = title_highlight[0] if title_highlight else None
description_html = description_highlight[0] if description_highlight else None
documents.append(
{
"video_id": source.get("video_id"),
"channel_id": source.get("channel_id"),
"channel_name": source.get("channel_name"),
"title": source.get("title"),
"titleHtml": title_html,
"description": source.get("description"),
"descriptionHtml": description_html,
"date": source.get("date"),
"duration": source.get("duration"),
"url": source.get("url"),
"toHighlight": transcript_highlight,
"highlightSource": {
"primary": bool(highlight_map.get("transcript_full")),
"secondary": bool(highlight_map.get("transcript_secondary_full")),
},
"internal_references_count": source.get("internal_references_count", 0),
"internal_references": source.get("internal_references", []),
"referenced_by_count": source.get("referenced_by_count", 0),
"referenced_by": source.get("referenced_by", []),
"video_status": source.get("video_status"),
"external_reference": source.get("external_reference"),
}
)
return jsonify(
{
"items": documents,
"totalResults": total,
"totalPages": (total + size - 1) // size,
"currentPage": page,
}
)
@app.route("/api/metrics")
def metrics():
include_external = request.args.get("external", default="1", type=str)
include_external = include_external.lower() not in {"0", "false", "no"}
try:
data = elastic_metrics_payload(
client,
index,
channel_field_candidates=["channel_id.keyword", "channel_id"],
debug=config.elastic.debug,
include_external=include_external,
)
except Exception:
LOGGER.exception(
"Falling back to local metrics payload due to Elasticsearch error.",
exc_info=True,
)
data = metrics_payload(config.data.root, include_external=include_external)
return jsonify(data)
@app.route("/api/frequency")
def frequency():
raw_term = request.args.get("term", type=str) or ""
use_query_string = request.args.get("query_string", default="0", type=str)
use_query_string = (use_query_string or "").lower() in {"1", "true", "yes"}
term = raw_term.strip()
if not term and not use_query_string:
return ("term parameter is required", 400)
if use_query_string and not term:
term = "*"
raw_channels: List[Optional[str]] = request.args.getlist("channel_id")
legacy_channel = request.args.get("channel", type=str)
if legacy_channel:
raw_channels.append(legacy_channel)
channels = parse_channel_params(raw_channels)
year = request.args.get("year", "", type=str) or None
interval = (request.args.get("interval", "month") or "month").lower()
allowed_intervals = {"day", "week", "month", "quarter", "year"}
if interval not in allowed_intervals:
interval = "month"
start = request.args.get("start", type=str)
end = request.args.get("end", type=str)
def parse_flag(name: str, default: bool = True) -> bool:
value = request.args.get(name)
if value is None:
return default
lowered = value.lower()
return lowered not in {"0", "false", "no"}
use_exact = parse_flag("exact", True)
use_fuzzy = parse_flag("fuzzy", True)
use_phrase = parse_flag("phrase", True)
include_external = parse_flag("external", False)
if use_query_string:
use_exact = use_fuzzy = use_phrase = False
search_payload = build_query_payload(
term,
channels=channels,
year=year,
sort="relevant",
use_exact=use_exact,
use_fuzzy=use_fuzzy,
use_phrase=use_phrase,
use_query_string=use_query_string,
include_external=include_external,
)
query = search_payload.get("query", {"match_all": {}})
if start or end:
range_filter: Dict[str, Dict[str, Dict[str, str]]] = {"range": {"date": {}}}
if start:
range_filter["range"]["date"]["gte"] = start
if end:
range_filter["range"]["date"]["lte"] = end
if "bool" in query:
bool_clause = query.setdefault("bool", {})
existing_filter = bool_clause.get("filter")
if existing_filter is None:
bool_clause["filter"] = [range_filter]
elif isinstance(existing_filter, list):
bool_clause["filter"].append(range_filter)
else:
bool_clause["filter"] = [existing_filter, range_filter]
elif query.get("match_all") is not None:
query = {"bool": {"filter": [range_filter]}}
else:
query = {"bool": {"must": [query], "filter": [range_filter]}}
histogram: Dict[str, Any] = {
"field": "date",
"calendar_interval": interval,
"min_doc_count": 0,
}
if start or end:
bounds: Dict[str, str] = {}
if start:
bounds["min"] = start
if end:
bounds["max"] = end
if bounds:
histogram["extended_bounds"] = bounds
channel_terms_size = max(6, len(channels)) if channels else 6
body = {
"size": 0,
"query": query,
"aggs": {
"over_time": {
"date_histogram": histogram,
"aggs": {
"by_channel": {
"terms": {
"field": "channel_id.keyword",
"size": channel_terms_size,
"order": {"_count": "desc"},
},
"aggs": {
"channel_name_hit": {
"top_hits": {
"size": 1,
"_source": {"includes": ["channel_name"]},
"sort": [
{
"channel_name.keyword": {
"order": "asc",
"missing": "_last",
"unmapped_type": "keyword",
}
}
],
}
}
},
}
},
}
},
}
if config.elastic.debug:
LOGGER.info(
"Elasticsearch frequency request: %s",
json.dumps(
{
"index": index,
"body": body,
"term": term,
"interval": interval,
"channels": channels,
"start": start,
"end": end,
"query_string": use_query_string,
},
indent=2,
),
)
response = client.search(index=index, body=body)
if config.elastic.debug:
LOGGER.info(
"Elasticsearch frequency response: %s",
json.dumps(response, indent=2, default=str),
)
raw_buckets = (
response.get("aggregations", {})
.get("over_time", {})
.get("buckets", [])
)
channel_totals: Dict[str, Dict[str, Any]] = {}
buckets: List[Dict[str, Any]] = []
for bucket in raw_buckets:
date_str = bucket.get("key_as_string")
total = bucket.get("doc_count", 0)
channel_entries: List[Dict[str, Any]] = []
for ch_bucket in bucket.get("by_channel", {}).get("buckets", []):
cid = ch_bucket.get("key")
count = ch_bucket.get("doc_count", 0)
if cid:
hit_source = (
ch_bucket.get("channel_name_hit", {})
.get("hits", {})
.get("hits", [{}])[0]
.get("_source", {})
)
channel_name = hit_source.get("channel_name") if isinstance(hit_source, dict) else None
channel_entries.append({"id": cid, "count": count, "name": channel_name})
if cid not in channel_totals:
channel_totals[cid] = {"total": 0, "name": channel_name}
channel_totals[cid]["total"] += count
if channel_name and not channel_totals[cid].get("name"):
channel_totals[cid]["name"] = channel_name
buckets.append(
{"date": date_str, "total": total, "channels": channel_entries}
)
ranked_channels = sorted(
[
{"id": cid, "total": info.get("total", 0), "name": info.get("name")}
for cid, info in channel_totals.items()
],
key=lambda item: item["total"],
reverse=True,
)
payload = {
"term": raw_term if not use_query_string else term,
"interval": interval,
"buckets": buckets,
"channels": ranked_channels,
"totalResults": response.get("hits", {})
.get("total", {})
.get("value", 0),
}
return jsonify(payload)
@app.route("/frequency")
def frequency_page():
return send_from_directory(app.static_folder, "frequency.html")
@app.route("/api/transcript")
def transcript():
video_id = request.args.get("video_id", type=str)
if not video_id:
return ("video_id not set", 400)
response = client.get(index=index, id=video_id, ignore=[404])
if config.elastic.debug:
LOGGER.info(
"Elasticsearch transcript request: index=%s id=%s", index, video_id
)
LOGGER.info(
"Elasticsearch transcript response: %s",
json.dumps(response, indent=2, default=str)
if response
else "None",
)
if not response or not response.get("found"):
return ("not found", 404)
source = response["_source"]
return jsonify(
{
"video_id": source.get("video_id"),
"title": source.get("title"),
"transcript_parts": source.get("transcript_parts", []),
"transcript_full": source.get("transcript_full"),
"transcript_secondary_parts": source.get("transcript_secondary_parts", []),
"transcript_secondary_full": source.get("transcript_secondary_full"),
}
)
return app
def main() -> None: # pragma: no cover
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
app = create_app()
debug_mode = os.environ.get("FLASK_DEBUG", "0").lower() in ("1", "true")
app.run(host="0.0.0.0", port=8080, debug=debug_mode)
if __name__ == "__main__": # pragma: no cover
main()