import re
import sqlite3
from difflib import SequenceMatcher
from datetime import datetime
from pathlib import Path
from config import DB_PATH, DATA_DIR, YOUTUBE_RESEARCH_DB_PATH
CONTENT_TYPES = [
"Muster", "Idee", "Hook", "Struktur", "Argument", "Anleitung",
"Beispiel", "Vorlage", "Checkliste", "Story", "Framework", "Umsetzungskandidat",
]
STATUS_VALUES = ["offen", "prüfen", "interessant", "umsetzen", "verwerfen"]
LEVEL_VALUES = ["niedrig", "mittel", "hoch"]
SCHEMA = """
CREATE TABLE IF NOT EXISTS ce_sources (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_system TEXT NOT NULL DEFAULT 'youtube-research',
source_video_id INTEGER NOT NULL,
youtube_id TEXT,
title TEXT,
channel TEXT,
language TEXT,
published_at TEXT,
youtube_url TEXT,
transcript_status TEXT,
obsidian_artifact_path TEXT,
graphiti_status TEXT,
graphiti_import_key TEXT,
imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
source_snapshot_hash TEXT,
UNIQUE(source_system, source_video_id)
);
CREATE TABLE IF NOT EXISTS ce_collections (
id INTEGER PRIMARY KEY AUTOINCREMENT,
slug TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
description TEXT,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS ce_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_id INTEGER NOT NULL REFERENCES ce_sources(id) ON DELETE CASCADE,
collection_id INTEGER REFERENCES ce_collections(id) ON DELETE SET NULL,
content_type TEXT NOT NULL DEFAULT 'unbestimmt',
is_candidate INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'offen',
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(source_id)
);
CREATE TABLE IF NOT EXISTS ce_ratings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id INTEGER NOT NULL REFERENCES ce_items(id) ON DELETE CASCADE,
nachahmbarkeit INTEGER NOT NULL DEFAULT 3 CHECK(nachahmbarkeit BETWEEN 1 AND 5),
aufwand TEXT NOT NULL DEFAULT 'mittel' CHECK(aufwand IN ('niedrig','mittel','hoch')),
umsetzbarkeit TEXT NOT NULL DEFAULT 'mittel' CHECK(umsetzbarkeit IN ('niedrig','mittel','hoch')),
erwarteter_nutzen TEXT NOT NULL DEFAULT 'mittel' CHECK(erwarteter_nutzen IN ('niedrig','mittel','hoch')),
rated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(item_id)
);
CREATE TABLE IF NOT EXISTS ce_notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id INTEGER NOT NULL REFERENCES ce_items(id) ON DELETE CASCADE,
note_text TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
"""
# Spalten, die nach dem urspruenglichen Grundgeruest-Schema ergaenzt wurden,
# um die vom read-only Import benoetigten Felder aufzunehmen.
_CE_SOURCES_MIGRATION_COLUMNS = {
"summary": "TEXT",
"skill_file": "TEXT",
"is_high_value": "INTEGER",
"claude_score": "INTEGER",
}
def get_db():
"""Verbindung zur eigenen content_extraction.db. Einzige Datei, in die geschrieben wird."""
DATA_DIR.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
return conn
def _migrate_ce_sources_columns(conn):
existing = {row[1] for row in conn.execute("PRAGMA table_info(ce_sources)").fetchall()}
for col, coltype in _CE_SOURCES_MIGRATION_COLUMNS.items():
if col not in existing:
conn.execute(f"ALTER TABLE ce_sources ADD COLUMN {col} {coltype}")
def _migrate_ce_ratings_schema(conn):
"""Aeltere Installationen hatten ce_ratings ohne erwarteter_nutzen und mit
nachahmbarkeit als Text-Enum statt Integer 1-5. Da ce_ratings zum Zeitpunkt
dieser Migration nachweislich leer war (0 Zeilen), wird die Tabelle sicher
per Drop+Recreate auf das neue Schema migriert. Enthaelt die Tabelle bereits
Zeilen, wird NICHT automatisch migriert (kein Datenverlust-Risiko eingehen)."""
existing = {row[1] for row in conn.execute("PRAGMA table_info(ce_ratings)").fetchall()}
if "erwarteter_nutzen" not in existing:
count = conn.execute("SELECT COUNT(*) FROM ce_ratings").fetchone()[0]
if count == 0:
conn.execute("DROP TABLE ce_ratings")
conn.executescript("""
CREATE TABLE ce_ratings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id INTEGER NOT NULL REFERENCES ce_items(id) ON DELETE CASCADE,
nachahmbarkeit INTEGER NOT NULL DEFAULT 3 CHECK(nachahmbarkeit BETWEEN 1 AND 5),
aufwand TEXT NOT NULL DEFAULT 'mittel' CHECK(aufwand IN ('niedrig','mittel','hoch')),
umsetzbarkeit TEXT NOT NULL DEFAULT 'mittel' CHECK(umsetzbarkeit IN ('niedrig','mittel','hoch')),
erwarteter_nutzen TEXT NOT NULL DEFAULT 'mittel' CHECK(erwarteter_nutzen IN ('niedrig','mittel','hoch')),
rated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(item_id)
);
""")
# sonst: bewusst nicht anfassen, Migration muesste dann manuell erfolgen
def init_db():
conn = get_db()
conn.executescript(SCHEMA)
_migrate_ce_sources_columns(conn)
_migrate_ce_ratings_schema(conn)
conn.commit()
conn.close()
def get_youtube_research_ro():
"""Read-only Verbindung zu YouTube Research knowledge.db.
Wird ausschliesslich lesend genutzt, niemals fuer Schreiboperationen."""
uri = f"file:{YOUTUBE_RESEARCH_DB_PATH}?mode=ro"
return sqlite3.connect(uri, uri=True)
def count_sources():
conn = get_db()
n = conn.execute("SELECT COUNT(*) FROM ce_sources").fetchone()[0]
conn.close()
return n
def count_items():
conn = get_db()
n = conn.execute("SELECT COUNT(*) FROM ce_items").fetchone()[0]
conn.close()
return n
# ---------------------------------------------------------------------------
# Such-/Filterfunktionen
# ---------------------------------------------------------------------------
# Alle Filter werden ausschliesslich per SQL-Parameterbindung (?) angewendet,
# nie per String-Konkatenation von Nutzereingaben. Einfache Ergebnisbegrenzung
# (MAX_RESULTS) statt vollstaendiger Pagination -- fuer die aktuelle Datenmenge
# (877 Quellen, wenige Kandidaten) ausreichend; echte Pagination (LIMIT/OFFSET
# plus Seiten-UI) ist ein moeglicher naechster Schritt, hier nur vorbereitet
# durch den zentralen MAX_RESULTS-Parameter.
MAX_RESULTS = 1000
SEARCH_FIELD_WEIGHTS = {
"title": 10.0,
"summary": 6.0,
"channel": 4.0,
"skill_file": 2.0,
"obsidian_artifact_path": 1.5,
"graphiti_import_key": 0.0,
}
SEARCH_THEME_EXPANSIONS = {
"social media": [
"social media", "social-media", "social media posts", "social-media-post-erstellung",
"posts", "post", "beiträge", "content", "content creation", "content-erstellung",
"linkedin", "instagram", "reels", "carousel", "hook", "hooks", "kurzvideo",
"kurzvideos", "shorts", "caption", "copy", "social content",
],
"social media posts": [
"social media", "social-media", "social media posts", "social-media-post-erstellung",
"posts", "post", "beiträge", "content", "content creation", "content-erstellung",
"linkedin", "instagram", "reels", "carousel", "hook", "hooks", "kurzvideo",
"kurzvideos", "shorts", "caption", "copy",
],
"content": [
"content", "content creation", "content-erstellung", "content erstellung",
"content strategie", "content-strategie", "content plan", "content-plan",
"content marketing", "social media", "linkedin", "instagram", "reels", "hook", "hooks",
],
"content creation": [
"content creation", "content-erstellung", "content erstellung", "content",
"social media posts", "linkedin", "instagram", "reels", "carousel", "hook", "hooks",
],
"linkedin": ["linkedin", "linkedin content", "linkedin posts", "beiträge", "posts", "post", "content"],
"instagram": ["instagram", "instagram content", "reels", "carousel", "posts", "post", "content"],
"reels": ["reels", "shorts", "kurzvideo", "kurzvideos", "video hooks", "hook", "hooks"],
"carousel": ["carousel", "karussell", "linkedin carousel", "instagram carousel", "posts", "content"],
"hook": ["hook", "hooks", "youtube hooks", "social hooks", "content hook", "title hook"],
"hooks": ["hook", "hooks", "youtube hooks", "social hooks", "content hook", "title hook"],
"kurzvideo": ["kurzvideo", "kurzvideos", "shorts", "reels", "social media posts", "hook"],
"kurzvideos": ["kurzvideo", "kurzvideos", "shorts", "reels", "social media posts", "hook"],
"post": ["post", "posts", "beiträge", "beitrag", "social media posts", "content", "linkedin", "instagram"],
"posts": ["post", "posts", "beiträge", "beitrag", "social media posts", "content", "linkedin", "instagram"],
"social": ["social media", "social media posts", "social-media", "content", "linkedin", "instagram"],
"media": ["social media", "social media posts", "content", "linkedin", "instagram", "reels", "hook"],
"social-media-post-erstellung": [
"social-media-post-erstellung", "social media posts", "content creation", "content-erstellung",
"linkedin", "instagram", "reels", "hook", "hooks", "carousel",
],
}
TAG_RULES = [
{"key": "social_media", "label": "Social Media", "needles": ["social media", "social-media", "social"]},
{"key": "social_media_posts", "label": "Social Media Posts", "needles": ["social media posts", "social-media-post-erstellung", "posts", "post"]},
{"key": "linkedin", "label": "LinkedIn", "needles": ["linkedin"]},
{"key": "instagram", "label": "Instagram", "needles": ["instagram"]},
{"key": "content", "label": "Content", "needles": ["content"]},
{"key": "content_marketing", "label": "Content Marketing", "needles": ["content marketing"]},
{"key": "post_erstellung", "label": "Post-Erstellung", "needles": ["post-erstellung", "post erstellung", "posts", "post", "beiträge", "beitrag"]},
{"key": "hook", "label": "Hook", "needles": ["hook", "hooks"]},
{"key": "reels", "label": "Reels", "needles": ["reels"]},
{"key": "carousel", "label": "Carousel", "needles": ["carousel"]},
{"key": "canva", "label": "Canva", "needles": ["canva"]},
{"key": "claude", "label": "Claude", "needles": ["claude"]},
{"key": "ki", "label": "KI", "needles": ["ki", "ai"]},
{"key": "automatisierung", "label": "Automatisierung", "needles": ["automatisierung", "automation", "automate"]},
{"key": "youtube", "label": "YouTube", "needles": ["youtube"]},
{"key": "shorts", "label": "Shorts", "needles": ["shorts"]},
{"key": "skript", "label": "Skript", "needles": ["skript", "script"]},
{"key": "agenten", "label": "Agenten", "needles": ["agenten", "agent"]},
{"key": "obsidian", "label": "Obsidian", "needles": ["obsidian"]},
{"key": "graphiti", "label": "Graphiti", "needles": ["graphiti"]},
{"key": "n8n", "label": "n8n", "needles": ["n8n"]},
{"key": "luftqualitat", "label": "Luftqualität", "needles": ["luftqualität", "luftqualitaet", "air quality", "airquality"]},
{"key": "feuchteanalyse", "label": "Feuchteanalyse", "needles": ["feuchteanalyse", "feuchte", "humidity"]},
{"key": "luftung", "label": "Lüftung", "needles": ["lüftung", "lueftung", "ventilation"]},
{"key": "kundenkommunikation", "label": "Kundenkommunikation", "needles": ["kundenkommunikation", "customer communication", "kommunikation"]},
]
TAG_RULE_LOOKUP = {}
def _tag_lookup_norm(value):
text = str(value or "").lower()
text = text.replace("-", " ").replace("_", " ").replace("/", " ")
text = re.sub(r"\s+", " ", text)
return text.strip()
for _rule in TAG_RULES:
TAG_RULE_LOOKUP[_rule["key"]] = _rule
TAG_RULE_LOOKUP[_tag_lookup_norm(_rule["label"])] = _rule
for _needle in _rule["needles"]:
TAG_RULE_LOOKUP[_tag_lookup_norm(_needle)] = _rule
SEARCH_RESULT_TERM_GROUPS = [
("Social Media", ["social media", "social-media", "social"]),
("Social Media Posts", ["social media posts", "social-media-post-erstellung", "posts", "post"]),
("LinkedIn", ["linkedin"]),
("Instagram", ["instagram"]),
("Content", ["content"]),
("Content Marketing", ["content marketing"]),
("Post-Erstellung", ["post-erstellung", "post erstellung", "posts", "post", "beiträge", "beitrag"]),
("Hook", ["hook", "hooks"]),
("Reels", ["reels"]),
("Carousel", ["carousel"]),
("Canva", ["canva"]),
("Claude", ["claude"]),
("KI", ["ki", "ai"]),
("Automatisierung", ["automatisierung", "automation", "automate"]),
("YouTube", ["youtube"]),
("Shorts", ["shorts"]),
("Skript", ["skript", "script"]),
("Agenten", ["agenten", "agent"]),
("Obsidian", ["obsidian"]),
("Graphiti", ["graphiti"]),
("n8n", ["n8n"]),
("Luftqualität", ["luftqualität", "luftqualitaet", "air quality", "airquality"]),
("Feuchteanalyse", ["feuchteanalyse", "feuchte", "humidity"]),
("Lüftung", ["lüftung", "lueftung", "ventilation"]),
("Kundenkommunikation", ["kundenkommunikation", "customer communication", "kommunikation"]),
]
MATCH_FIELD_LABELS = {
"title": "Titel",
"summary": "Zusammenfassung",
"channel": "Kanal",
"youtube_id": "YouTube-ID",
"obsidian_artifact_path": "Artefakt-Dateiname",
"skill_file": "Skill-Dateiname",
"graphiti_import_key": "Graphiti",
"high_value": "High Value",
}
def _normalize_search_text(text):
text = (text or "").lower()
text = text.replace("-", " ").replace("_", " ").replace("/", " ")
text = re.sub(r"\s+", " ", text)
return text.strip()
def _filename_text(value):
if not value:
return ""
try:
return Path(str(value)).name
except Exception:
return str(value)
def _searchable_path_text(value):
if not value:
return ""
filename = _filename_text(value)
if not filename:
return ""
return _normalize_search_text(filename)
def _searchable_content_text(row):
parts = [
row.get("title"),
row.get("summary"),
row.get("channel"),
_filename_text(row.get("skill_file")),
_filename_text(row.get("obsidian_artifact_path")),
]
return _normalize_search_text(" ".join(str(part or "") for part in parts))
def _expand_search_terms(query):
normalized = _normalize_search_text(query)
if not normalized:
return []
terms = []
seen = set()
def add(term):
term = _normalize_search_text(term)
if term and term not in seen:
seen.add(term)
terms.append(term)
add(normalized)
for token in normalized.split(" "):
add(token)
for phrase, expansions in SEARCH_THEME_EXPANSIONS.items():
if phrase in normalized:
add(phrase)
for exp in expansions:
add(exp)
return sorted(terms, key=len, reverse=True)
def _unique_preserve_order(values):
out = []
seen = set()
for value in values:
if value not in seen:
seen.add(value)
out.append(value)
return out
def _text_contains_any(text, needles):
text = _normalize_search_text(text)
return any(needle and needle in text for needle in needles)
def _collect_result_groups(row, groups):
combined = _searchable_content_text(row)
hits = []
for label, needles in groups:
if any(needle in combined for needle in needles):
hits.append(label)
return hits
def _build_visible_tags(row, matched_terms=None, limit=8):
tag_rules = [
("Social Media", ["social media", "social-media", "social"]),
("LinkedIn", ["linkedin"]),
("Instagram", ["instagram"]),
("Content Marketing", ["content marketing", "content-strategie", "content strategie"]),
("Content", ["content"]),
("Post-Erstellung", ["post-erstellung", "post erstellung", "posts", "post", "beiträge", "beitrag"]),
("Hook", ["hook", "hooks"]),
("Reels", ["reels"]),
("Carousel", ["carousel"]),
("Canva", ["canva"]),
("Claude", ["claude"]),
("KI", ["ki", "ai"]),
("Automatisierung", ["automatisierung", "automation", "automate"]),
("YouTube", ["youtube"]),
("Shorts", ["shorts"]),
("Skript", ["skript", "script"]),
("Agenten", ["agenten", "agent"]),
("Obsidian", ["obsidian"]),
("Graphiti", ["graphiti"]),
("n8n", ["n8n"]),
("Luftqualität", ["luftqualität", "luftqualitaet", "air quality", "airquality"]),
("Feuchteanalyse", ["feuchteanalyse", "feuchte", "humidity"]),
("Lüftung", ["lüftung", "lueftung", "ventilation"]),
("Kundenkommunikation", ["kundenkommunikation", "customer communication", "kommunikation"]),
("Mallorca AirServices", ["mallorca airservices"]),
("LüftungsProfi", ["lüftungsprofi", "lueftungsprofi"]),
]
combined = _normalize_search_text(" ".join(str(row.get(field) or "") for field in (
"title", "summary", "channel", "obsidian_artifact_path",
"skill_file", "graphiti_import_key", "youtube_id", "youtube_url",
)))
matched_terms = [t for t in (matched_terms or []) if t]
tags = []
seen = set()
for label, needles in tag_rules:
if any(needle in combined for needle in needles) or any(needle in _normalize_search_text(" ".join(matched_terms)) for needle in needles):
if label not in seen:
tags.append(label)
seen.add(label)
if len(tags) >= limit:
break
return tags[:limit]
def _search_excerpt(text, terms, limit=220):
text = text or ""
lowered = text.lower()
best_pos = None
best_term = None
for term in terms:
pos = lowered.find(term)
if pos != -1 and (best_pos is None or pos < best_pos):
best_pos = pos
best_term = term
if best_pos is None:
excerpt = text[:limit].strip()
return excerpt + ("…" if len(text) > limit else "")
start = max(0, best_pos - 70)
end = min(len(text), best_pos + max(len(best_term), 90))
excerpt = text[start:end].strip()
if start > 0:
excerpt = "…" + excerpt
if end < len(text):
excerpt = excerpt + "…"
return excerpt
def _score_source_row(row, query, terms):
haystacks = {
"title": _normalize_search_text(row.get("title")),
"channel": _normalize_search_text(row.get("channel")),
"summary": _normalize_search_text(row.get("summary")),
"obsidian_artifact_path": _searchable_path_text(row.get("obsidian_artifact_path")),
"skill_file": _searchable_path_text(row.get("skill_file")),
}
score = 0.0
matched_terms = []
matched_fields = []
query_norm = _normalize_search_text(query)
for field, text in haystacks.items():
if not text:
continue
field_match = False
for term in terms:
if term and term in text:
score += SEARCH_FIELD_WEIGHTS[field]
matched_terms.append(term)
field_match = True
if query_norm and query_norm in text:
score += SEARCH_FIELD_WEIGHTS[field] * 1.5
matched_terms.append(query_norm)
field_match = True
# Kontexttoleranter Einzelwort-Match: Ein laengeres Suchwort mit
# minimalem Tippfehler darf einen passenden Titel/Summary-Treffer
# nicht ausblenden. Die Schwelle ist bewusst streng, damit keine
# beliebigen unscharfen Treffer entstehen.
if not field_match and len(query_norm) >= 6 and ' ' not in query_norm:
candidates = set(re.findall(r'[a-z0-9äöüß]+', text))
for candidate in candidates:
if len(candidate) < 6 or abs(len(candidate) - len(query_norm)) > 2:
continue
similarity = SequenceMatcher(None, query_norm, candidate).ratio()
if similarity >= 0.88:
score += SEARCH_FIELD_WEIGHTS[field] * 0.9
matched_terms.append(candidate)
field_match = True
break
if field_match:
matched_fields.append(field)
if row.get("is_high_value"):
score += 5.0
matched_fields.append("high_value")
if row.get("claude_score") is not None:
try:
score += min(float(row.get("claude_score") or 0) / 10.0, 2.0)
except (TypeError, ValueError):
pass
matched_terms = _unique_preserve_order(matched_terms)
matched_fields = _unique_preserve_order(matched_fields)
return score, matched_terms, matched_fields
def _format_match_reason(matched_fields):
labels = []
for field in matched_fields:
label = MATCH_FIELD_LABELS.get(field)
if label and label not in labels:
labels.append(label)
if not labels:
return "Begriffs-Match"
return ", ".join(labels[:4])
def _query_core_terms(query):
normalized = _normalize_search_text(query)
if not normalized:
return []
terms = []
seen = set()
def add(term):
term = _normalize_search_text(term)
if term and term not in seen:
seen.add(term)
terms.append(term)
add(normalized)
for token in normalized.split():
add(token)
return terms
def _normalize_filter_tag(tag):
tag_norm = _normalize_search_text(tag)
rule = TAG_RULE_LOOKUP.get(tag_norm)
return rule["key"] if rule else tag_norm.replace(" ", "_")
def _resolve_tag_rule(tag):
if not tag:
return None
return TAG_RULE_LOOKUP.get(_normalize_search_text(tag)) or TAG_RULE_LOOKUP.get(tag)
def _build_visible_tags(row, matched_terms=None, limit=8):
combined = _searchable_content_text(row)
matched_terms_text = _normalize_search_text(" ".join(t for t in (matched_terms or []) if t))
tags = []
seen = set()
for rule in TAG_RULES:
if any(needle in combined for needle in rule["needles"]) or any(needle in matched_terms_text for needle in rule["needles"]):
if rule["key"] not in seen:
tags.append({"key": rule["key"], "label": rule["label"]})
seen.add(rule["key"])
if len(tags) >= limit:
break
return tags[:limit]
def _row_tag_keys(row, matched_terms=None):
return [tag["key"] for tag in _build_visible_tags(row, matched_terms, limit=len(TAG_RULES))]
def _row_has_tag_key(row, tag_key, matched_terms=None):
normalized_tag_key = _normalize_filter_tag(tag_key)
return normalized_tag_key in _row_tag_keys(row, matched_terms)
def _classify_match_quality(query, matched_terms):
core_terms = set(_query_core_terms(query))
for term in matched_terms or []:
if _normalize_search_text(term) in core_terms:
return "eng"
return "broad"
def _format_match_quality_label(match_quality):
return "Enger Treffer" if match_quality == "eng" else "Erweiterter Treffer"
def _search_excerpt(text, terms, limit=220):
text = text or ""
lowered = text.lower()
best_pos = None
best_term = None
for term in terms:
pos = lowered.find(term)
if pos != -1 and (best_pos is None or pos < best_pos):
best_pos = pos
best_term = term
if best_pos is None:
excerpt = text[:limit].strip()
return excerpt + ("…" if len(text) > limit else "")
start = max(0, best_pos - 70)
end = min(len(text), best_pos + max(len(best_term), 90))
excerpt = text[start:end].strip()
if start > 0:
excerpt = "…" + excerpt
if end < len(text):
excerpt = excerpt + "…"
return excerpt
def _parse_source_datetime(value):
if not value:
return None
text = str(value).strip()
if not text:
return None
candidates = [text, text.replace('Z', '+00:00')]
for candidate in candidates:
try:
return datetime.fromisoformat(candidate)
except ValueError:
pass
for fmt in (
'%Y-%m-%d %H:%M:%S',
'%Y-%m-%d %H:%M',
'%Y-%m-%d',
'%d.%m.%Y %H:%M:%S',
'%d.%m.%Y %H:%M',
'%d.%m.%Y',
):
try:
return datetime.strptime(text, fmt)
except ValueError:
pass
return None
def _row_sort_key_newest(row):
published = _parse_source_datetime(row.get('published_at'))
imported = _parse_source_datetime(row.get('imported_at'))
best = published or imported
ts = best.timestamp() if best else 0.0
return (ts, int(row.get('source_video_id') or 0), int(row.get('id') or 0))
def _row_sort_key_score(row):
published = _parse_source_datetime(row.get('published_at'))
imported = _parse_source_datetime(row.get('imported_at'))
best = published or imported
ts = best.timestamp() if best else 0.0
return (-float(row.get('search_score') or 0.0), -ts, -int(row.get('source_video_id') or 0), -int(row.get('id') or 0))
def search_videos(query, tag=None, limit=50, sort='score'):
terms = _expand_search_terms(query)
active_rule = _resolve_tag_rule(tag)
active_tag_key = active_rule["key"] if active_rule else _normalize_filter_tag(tag)
if not terms:
return [], {"query": query, "active_tag": active_tag_key, "active_tag_label": active_rule["label"] if active_rule else (tag or ""), "expanded_terms": [], "per_term_counts": [], "total_count": 0, "shown_count": 0, "has_more": False, "limit": limit, "total_matches": 0, "returned_count": 0, "sort": sort, "narrow_count": 0, "broad_count": 0, "base_total_count": 0, "limit_param": str(limit)}
conn = get_db()
try:
rows = conn.execute("SELECT * FROM ce_sources").fetchall()
finally:
conn.close()
base_scored = []
query_norm = _normalize_search_text(query)
for row in rows:
row = dict(row)
score, matched_terms, matched_fields = _score_source_row(row, query, terms)
if score <= 0:
continue
summary = row.get("summary") or ""
youtube_id = (row.get("youtube_id") or "").strip()
visible_tags = _build_visible_tags(row, matched_terms, limit=8)
tag_keys = [t["key"] for t in visible_tags]
match_quality = _classify_match_quality(query, matched_terms)
row["search_score"] = round(score, 2)
row["match_terms"] = matched_terms[:12]
row["match_fields"] = matched_fields
row["match_reason"] = _format_match_reason(matched_fields)
row["match_quality"] = match_quality
row["match_quality_label"] = _format_match_quality_label(match_quality)
row["summary_excerpt"] = _search_excerpt(summary, terms) if summary.strip() else "Keine Zusammenfassung vorhanden"
row["detail_url"] = f"/videos/{row['id']}"
row["thumbnail_url"] = f"https://img.youtube.com/vi/{youtube_id}/hqdefault.jpg" if youtube_id else ""
row["visible_tags"] = visible_tags
row["tag_keys"] = tag_keys
base_scored.append(row)
sort = (sort or 'score').strip().lower()
if sort == 'newest':
base_scored.sort(key=_row_sort_key_newest, reverse=True)
else:
base_scored.sort(key=_row_sort_key_score)
per_term_counts = []
for rule in TAG_RULES:
count = sum(1 for row in base_scored if rule["key"] in row.get("tag_keys", []))
if count:
per_term_counts.append({"key": rule["key"], "label": rule["label"], "count": count})
if active_rule:
filtered = [row for row in base_scored if active_rule["key"] in row.get("tag_keys", [])]
elif active_tag_key:
filtered = [row for row in base_scored if active_tag_key in row.get("tag_keys", [])]
else:
filtered = list(base_scored)
total_count = len(filtered)
narrow_count = sum(1 for row in filtered if row.get("match_quality") == "eng")
broad_count = total_count - narrow_count
sliced = filtered[:limit]
return sliced, {
"query": query,
"active_tag": active_tag_key,
"active_tag_label": active_rule["label"] if active_rule else (tag or ""),
"expanded_terms": terms,
"per_term_counts": per_term_counts,
"total_count": total_count,
"shown_count": len(sliced),
"has_more": total_count > limit,
"limit": limit,
"limit_param": str(limit),
"total_matches": total_count,
"returned_count": len(sliced),
"sort": sort,
"narrow_count": narrow_count,
"broad_count": broad_count,
"base_total_count": len(base_scored),
}
def search_sources(query, limit=50):
return search_videos(query, limit=limit)
def _yn_to_bool(value):
"""Wandelt 'yes'/'no' (oder leer) aus Query-Parametern in True/False/None."""
if value in ("yes", "ja", "1", "true"):
return True
if value in ("no", "nein", "0", "false"):
return False
return None
def list_sources(filters=None):
filters = filters or {}
where, params = [], []
q = (filters.get("q") or "").strip()
if q:
like = f"%{q}%"
where.append(
"(title LIKE ? OR channel LIKE ? OR summary LIKE ? OR youtube_id LIKE ?)"
)
params.extend([like, like, like, like])
channel = (filters.get("channel") or "").strip()
if channel:
where.append("channel = ?")
params.append(channel)
language = (filters.get("language") or "").strip()
if language:
where.append("language = ?")
params.append(language)
transcript_status = (filters.get("transcript_status") or "").strip()
if transcript_status:
where.append("transcript_status = ?")
params.append(transcript_status)
is_high_value = _yn_to_bool(filters.get("is_high_value"))
if is_high_value is not None:
where.append("is_high_value = ?")
params.append(1 if is_high_value else 0)
has_artifact_path = _yn_to_bool(filters.get("has_artifact_path"))
if has_artifact_path is True:
where.append("obsidian_artifact_path IS NOT NULL AND obsidian_artifact_path != ''")
elif has_artifact_path is False:
where.append("(obsidian_artifact_path IS NULL OR obsidian_artifact_path = '')")
has_item = _yn_to_bool(filters.get("has_item"))
if has_item is True:
where.append("id IN (SELECT source_id FROM ce_items)")
elif has_item is False:
where.append("id NOT IN (SELECT source_id FROM ce_items)")
sql = "SELECT * FROM ce_sources"
if where:
sql += " WHERE " + " AND ".join(where)
sql += " ORDER BY id LIMIT ?"
params.append(MAX_RESULTS)
conn = get_db()
rows = conn.execute(sql, params).fetchall()
total = conn.execute(
"SELECT COUNT(*) FROM ce_sources" + (" WHERE " + " AND ".join(where) if where else ""),
params[:-1],
).fetchone()[0]
conn.close()
return [dict(r) for r in rows], total
def list_items(filters=None):
filters = filters or {}
where, params = [], []
q = (filters.get("q") or "").strip()
if q:
like = f"%{q}%"
where.append("(s.title LIKE ? OR s.channel LIKE ? OR s.summary LIKE ?)")
params.extend([like, like, like])
content_type = (filters.get("content_type") or "").strip()
if content_type:
where.append("i.content_type = ?")
params.append(content_type)
status = (filters.get("status") or "").strip()
if status:
where.append("i.status = ?")
params.append(status)
imitation_score = (filters.get("imitation_score") or "").strip()
if imitation_score:
where.append("r.nachahmbarkeit = ?")
params.append(int(imitation_score))
effort = (filters.get("effort") or "").strip()
if effort:
where.append("r.aufwand = ?")
params.append(effort)
feasibility = (filters.get("feasibility") or "").strip()
if feasibility:
where.append("r.umsetzbarkeit = ?")
params.append(feasibility)
expected_value = (filters.get("expected_value") or "").strip()
if expected_value:
where.append("r.erwarteter_nutzen = ?")
params.append(expected_value)
base = (
"FROM ce_items i "
"JOIN ce_sources s ON s.id = i.source_id "
"LEFT JOIN ce_ratings r ON r.item_id = i.id"
)
sql = (
"SELECT i.*, s.title AS source_title, s.channel AS source_channel, "
"r.nachahmbarkeit, r.aufwand, r.umsetzbarkeit, r.erwarteter_nutzen " + base
)
if where:
sql += " WHERE " + " AND ".join(where)
sql += " ORDER BY i.id LIMIT ?"
params.append(MAX_RESULTS)
conn = get_db()
rows = conn.execute(sql, params).fetchall()
total = conn.execute(
"SELECT COUNT(*) " + base + (" WHERE " + " AND ".join(where) if where else ""),
params[:-1],
).fetchone()[0]
conn.close()
return [dict(r) for r in rows], total
def get_distinct_source_values():
"""Liefert distincte Werte fuer Filter-Dropdowns auf /sources."""
conn = get_db()
channels = [r[0] for r in conn.execute(
"SELECT DISTINCT channel FROM ce_sources WHERE channel IS NOT NULL ORDER BY channel"
).fetchall()]
languages = [r[0] for r in conn.execute(
"SELECT DISTINCT language FROM ce_sources WHERE language IS NOT NULL ORDER BY language"
).fetchall()]
statuses = [r[0] for r in conn.execute(
"SELECT DISTINCT transcript_status FROM ce_sources WHERE transcript_status IS NOT NULL ORDER BY transcript_status"
).fetchall()]
conn.close()
return {"channels": channels, "languages": languages, "transcript_statuses": statuses}
# ---------------------------------------------------------------------------
# Read-only Import aus YouTube Research (knowledge.db)
# ---------------------------------------------------------------------------
# get_youtube_research_ro() liefert eine ausschliesslich lesende Verbindung
# (SQLite URI mode=ro). Es wird an keiner Stelle in knowledge.db geschrieben.
# Keine Fundus-Tabellen (fundus_collections, fundus_items) werden gelesen.
_IMPORT_SOURCE_QUERY = """
SELECT
v.id AS source_video_id,
v.youtube_id AS youtube_id,
v.title AS title,
v.channel AS channel,
v.language AS language,
v.transcript_status AS transcript_status,
v.summary AS summary,
v.skill_file AS skill_file,
v.is_high_value AS is_high_value,
v.claude_score AS claude_score,
r.artifact_path AS artifact_path,
r.graphiti_status AS graphiti_status,
r.graphiti_import_key AS import_key
FROM videos v
LEFT JOIN source_processing_registry r
ON r.source_system = 'youtube-research'
AND r.source_id = CAST(v.id AS TEXT)
"""
_last_import_stats = {
"ran_at": None,
"read_count": 0,
"inserted": 0,
"updated": 0,
"unchanged": 0,
"error": None,
}
def get_last_import_stats():
return dict(_last_import_stats)
def import_sources_from_youtube_research():
"""Liest videos + source_processing_registry read-only aus knowledge.db
und spiegelt sie idempotent (Upsert ueber UNIQUE(source_system, source_video_id))
in ce_sources. Schreibt ausschliesslich in content_extraction.db."""
import datetime
stats = {
"ran_at": datetime.datetime.utcnow().isoformat() + "Z",
"read_count": 0,
"inserted": 0,
"updated": 0,
"unchanged": 0,
"error": None,
}
ro_conn = get_youtube_research_ro()
try:
ro_conn.row_factory = sqlite3.Row
rows = [dict(r) for r in ro_conn.execute(_IMPORT_SOURCE_QUERY).fetchall()]
finally:
ro_conn.close()
stats["read_count"] = len(rows)
conn = get_db()
try:
for row in rows:
youtube_url = (
f"https://youtube.com/watch?v={row['youtube_id']}"
if row["youtube_id"] else None
)
existing = conn.execute(
"""
SELECT id, youtube_id, title, channel, language, transcript_status,
summary, skill_file, is_high_value, claude_score,
obsidian_artifact_path, graphiti_status, graphiti_import_key
FROM ce_sources
WHERE source_system = 'youtube-research' AND source_video_id = ?
""",
(row["source_video_id"],),
).fetchone()
insert_values = (
"youtube-research",
row["source_video_id"],
row["youtube_id"],
row["title"],
row["channel"],
row["language"],
youtube_url,
row["transcript_status"],
row["summary"],
row["skill_file"],
row["is_high_value"],
row["claude_score"],
row["artifact_path"],
row["graphiti_status"],
row["import_key"],
)
if existing is None:
conn.execute(
"""
INSERT INTO ce_sources (
source_system, source_video_id, youtube_id, title, channel,
language, youtube_url, transcript_status, summary, skill_file,
is_high_value, claude_score, obsidian_artifact_path,
graphiti_status, graphiti_import_key
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
insert_values,
)
stats["inserted"] += 1
else:
changed = (
existing["youtube_id"] != row["youtube_id"]
or existing["title"] != row["title"]
or existing["channel"] != row["channel"]
or existing["language"] != row["language"]
or existing["transcript_status"] != row["transcript_status"]
or existing["summary"] != row["summary"]
or existing["skill_file"] != row["skill_file"]
or existing["is_high_value"] != row["is_high_value"]
or existing["claude_score"] != row["claude_score"]
or (row["artifact_path"] is not None and existing["obsidian_artifact_path"] != row["artifact_path"])
or existing["graphiti_status"] != row["graphiti_status"]
or existing["graphiti_import_key"] != row["import_key"]
)
if changed:
conn.execute(
"""
UPDATE ce_sources SET
youtube_id = ?, title = ?, channel = ?, language = ?,
youtube_url = ?, transcript_status = ?, summary = ?,
skill_file = ?, is_high_value = ?, claude_score = ?,
obsidian_artifact_path = ?, graphiti_status = ?,
graphiti_import_key = ?
WHERE id = ?
""",
(
row["youtube_id"], row["title"], row["channel"], row["language"],
youtube_url, row["transcript_status"], row["summary"],
row["skill_file"], row["is_high_value"], row["claude_score"],
row["artifact_path"] if row["artifact_path"] is not None else existing["obsidian_artifact_path"], row["graphiti_status"], row["import_key"],
existing["id"],
),
)
stats["updated"] += 1
else:
stats["unchanged"] += 1
conn.commit()
except Exception as exc:
conn.rollback()
stats["error"] = str(exc)
raise
finally:
conn.close()
_last_import_stats.update(stats)
return stats
# ---------------------------------------------------------------------------
# Manueller Content-Extraction-Workflow: Kandidat erzeugen + bewerten
# ---------------------------------------------------------------------------
# Kein Schreibzugriff auf knowledge.db, kein Graphiti-/Obsidian-Schreibzugriff,
# keine automatische KI-Extraktion -- alle Werte werden manuell uebergeben.
def get_source(source_id):
conn = get_db()
row = conn.execute("SELECT * FROM ce_sources WHERE id = ?", (source_id,)).fetchone()
conn.close()
return dict(row) if row else None
def create_item_from_source(source_id, content_type="unbestimmt"):
"""Erzeugt einen ce_item-Kandidaten aus einer ce_source.
Gibt (item_dict, created_bool) zurueck. Existiert bereits ein Kandidat fuer
diese Quelle (UNIQUE(source_id)), wird der bestehende Kandidat zurueckgegeben
und created=False gesetzt -- es entsteht kein Duplikat."""
if content_type not in CONTENT_TYPES:
content_type = "unbestimmt"
conn = get_db()
try:
source = conn.execute("SELECT id FROM ce_sources WHERE id = ?", (source_id,)).fetchone()
if source is None:
return None, False
existing = conn.execute(
"SELECT * FROM ce_items WHERE source_id = ?", (source_id,)
).fetchone()
if existing is not None:
return dict(existing), False
conn.execute(
"INSERT INTO ce_items (source_id, content_type, is_candidate, status) "
"VALUES (?, ?, 1, 'offen')",
(source_id, content_type),
)
conn.commit()
new_row = conn.execute(
"SELECT * FROM ce_items WHERE source_id = ?", (source_id,)
).fetchone()
return dict(new_row), True
finally:
conn.close()
def get_item_detail(item_id):
conn = get_db()
try:
item = conn.execute("SELECT * FROM ce_items WHERE id = ?", (item_id,)).fetchone()
if item is None:
return None
item = dict(item)
source = conn.execute(
"SELECT * FROM ce_sources WHERE id = ?", (item["source_id"],)
).fetchone()
rating = conn.execute(
"SELECT * FROM ce_ratings WHERE item_id = ?", (item_id,)
).fetchone()
notes = conn.execute(
"SELECT * FROM ce_notes WHERE item_id = ? ORDER BY created_at DESC", (item_id,)
).fetchall()
item["source"] = dict(source) if source else None
item["rating"] = dict(rating) if rating else None
item["notes"] = [dict(n) for n in notes]
return item
finally:
conn.close()
def save_item_rating(item_id, content_type=None, status=None, nachahmbarkeit=None,
aufwand=None, umsetzbarkeit=None, erwarteter_nutzen=None,
rationale=None):
"""Speichert Content-Typ, Status, Rating und Begruendung fuer einen Kandidaten.
Alle Parameter ausser item_id sind optional; nur uebergebene Felder werden
gesetzt bzw. mit sinnvollem Default bewertet. Wirft ValueError bei ungueltigen
Werten (kein stiller Fallback auf Defaults bei explizit falscher Eingabe)."""
conn = get_db()
try:
item = conn.execute("SELECT id FROM ce_items WHERE id = ?", (item_id,)).fetchone()
if item is None:
return None
if content_type is not None and content_type not in CONTENT_TYPES:
raise ValueError(f"Ungueltiger content_type: {content_type!r}")
if status is not None and status not in STATUS_VALUES:
raise ValueError(f"Ungueltiger status: {status!r}")
if content_type is not None or status is not None:
fields, values = [], []
if content_type is not None:
fields.append("content_type = ?")
values.append(content_type)
if status is not None:
fields.append("status = ?")
values.append(status)
fields.append("updated_at = CURRENT_TIMESTAMP")
values.append(item_id)
conn.execute(f"UPDATE ce_items SET {', '.join(fields)} WHERE id = ?", values)
n = nachahmbarkeit if nachahmbarkeit is not None else 3
a = aufwand if aufwand is not None else "mittel"
u = umsetzbarkeit if umsetzbarkeit is not None else "mittel"
e = erwarteter_nutzen if erwarteter_nutzen is not None else "mittel"
n = int(n)
if not (1 <= n <= 5):
raise ValueError("nachahmbarkeit muss zwischen 1 und 5 liegen")
for label, val in (("aufwand", a), ("umsetzbarkeit", u), ("erwarteter_nutzen", e)):
if val not in LEVEL_VALUES:
raise ValueError(f"{label} muss einer von {LEVEL_VALUES} sein, war {val!r}")
existing_rating = conn.execute(
"SELECT id FROM ce_ratings WHERE item_id = ?", (item_id,)
).fetchone()
if existing_rating:
conn.execute(
"UPDATE ce_ratings SET nachahmbarkeit=?, aufwand=?, umsetzbarkeit=?, "
"erwarteter_nutzen=?, rated_at=CURRENT_TIMESTAMP WHERE item_id=?",
(n, a, u, e, item_id),
)
else:
conn.execute(
"INSERT INTO ce_ratings (item_id, nachahmbarkeit, aufwand, "
"umsetzbarkeit, erwarteter_nutzen) VALUES (?, ?, ?, ?, ?)",
(item_id, n, a, u, e),
)
if rationale:
conn.execute(
"INSERT INTO ce_notes (item_id, note_text) VALUES (?, ?)",
(item_id, rationale),
)
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
return get_item_detail(item_id)
# ---------------------------------------------------------------------------
# Content-Matrix (read-only Auswertungsansicht)
# ---------------------------------------------------------------------------
# Reine SELECT-Abfrage ueber ce_items/ce_sources/ce_ratings. Kein INSERT,
# UPDATE oder DELETE. Erzeugt keine neuen Daten, keine automatische Bewertung.
def list_content_matrix():
"""Liest alle ce_items mit Quell- und Ratingdaten fuer die Matrix-Ansicht.
Sortierung: Nachahmbarkeit absteigend, dann erwarteter Nutzen (hoch>mittel>
niedrig), dann Umsetzbarkeit (hoch>mittel>niedrig), dann Aufwand (niedrig>
mittel>hoch). Items ohne Rating erscheinen am Ende (NULL sortiert zuletzt)."""
conn = get_db()
try:
rows = conn.execute(
"""
SELECT
i.id AS item_id,
i.content_type AS content_type,
i.status AS status,
s.title AS title,
s.channel AS channel,
s.youtube_id AS youtube_id,
s.youtube_url AS youtube_url,
r.nachahmbarkeit AS nachahmbarkeit,
r.aufwand AS aufwand,
r.umsetzbarkeit AS umsetzbarkeit,
r.erwarteter_nutzen AS erwarteter_nutzen
FROM ce_items i
JOIN ce_sources s ON s.id = i.source_id
LEFT JOIN ce_ratings r ON r.item_id = i.id
ORDER BY
r.nachahmbarkeit DESC,
CASE r.erwarteter_nutzen
WHEN 'hoch' THEN 1 WHEN 'mittel' THEN 2 WHEN 'niedrig' THEN 3 ELSE 4
END,
CASE r.umsetzbarkeit
WHEN 'hoch' THEN 1 WHEN 'mittel' THEN 2 WHEN 'niedrig' THEN 3 ELSE 4
END,
CASE r.aufwand
WHEN 'niedrig' THEN 1 WHEN 'mittel' THEN 2 WHEN 'hoch' THEN 3 ELSE 4
END,
i.id
"""
).fetchall()
return [dict(r) for r in rows]
finally:
conn.close()