#!/usr/bin/env python3
"""
Vollständige Audit-Prüfung aller 52 Obsidian-Dubletten
Erzeugt maschinenlesbaren JSON-Bericht und lesbaren Markdown-Bericht
Keine Änderungen an Dateien oder Datenbanken.
"""
import json
import hashlib
import os
import sqlite3
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Any, Optional
# === KONFIGURATION ===
KNOWLEDGE_DB = "/opt/struktur/youtube-research/knowledge.db"
CE_DB = "/opt/struktur/content-extraction/data/content_extraction.db"
OBSIDIAN_VAULT = "/opt/obsidian-vault/YouTube-Research"
CANONICAL_SCRIPT = "/opt/struktur/youtube-research/youtube_status_canonical.py"
REPORT_JSON = "/opt/struktur/reports/youtube-research-obsidian-dubletten-rohdaten.json"
REPORT_MD = "/opt/struktur/reports/youtube-research-obsidian-dubletten-verifiziert.md"
# === HILFSFUNKTIONEN ===
def run_canonical_audit() -> Dict:
"""Führt das kanonische Audit-Skript aus und gibt JSON zurück."""
result = subprocess.run(
[sys.executable, CANONICAL_SCRIPT],
capture_output=True, text=True, timeout=120
)
if result.returncode != 0:
raise RuntimeError(f"Canonical audit failed: {result.stderr}")
output = result.stdout
json_start = output.find("JSON OUTPUT:")
if json_start == -1:
raise RuntimeError("JSON OUTPUT marker not found in canonical audit output")
json_str = output[json_start + len("JSON OUTPUT:"):].strip()
first_brace = json_str.find('{')
if first_brace != -1:
json_str = json_str[first_brace:]
return json.loads(json_str)
def sha256_file(path: str) -> Optional[str]:
"""Berechnet SHA256 einer Datei über sha256sum."""
try:
result = subprocess.run(
["sha256sum", path],
capture_output=True, text=True, timeout=30
)
if result.returncode == 0:
return result.stdout.split()[0]
except Exception:
pass
return None
def file_exists(path: str) -> bool:
"""Prüft Dateiexistenz mit test -f."""
result = subprocess.run(["test", "-f", path], capture_output=True)
return result.returncode == 0
def get_file_stats(path: str) -> Dict:
"""Liefert Größe, mtime, Existenz, SHA256, Länge, Zeilen."""
exists = file_exists(path)
if not exists:
return {"exists": False, "size": None, "mtime": None, "sha256": None, "length": 0, "lines": 0}
stat = os.stat(path)
sha = sha256_file(path)
length = 0
lines = 0
try:
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
length = len(content)
lines = content.count('\n') + 1
except Exception:
pass
return {
"exists": True,
"size": stat.st_size,
"mtime": datetime.fromtimestamp(stat.st_mtime).isoformat(),
"sha256": sha,
"length": length,
"lines": lines
}
def read_file_content(path: str) -> str:
"""Liest Dateiinhalt sicher."""
try:
with open(path, 'r', encoding='utf-8') as f:
return f.read()
except Exception:
return ""
def parse_frontmatter(content: str) -> Dict:
"""Extrahiert Frontmatter (YAML zwischen ---)."""
if content.startswith('---'):
end = content.find('---', 3)
if end != -1:
fm_text = content[3:end].strip()
result = {}
for line in fm_text.split('\n'):
if ':' in line:
key, val = line.split(':', 1)
result[key.strip()] = val.strip()
return result
return {}
def extract_sections(content: str) -> Dict[str, str]:
"""Extrahiert Markdown-Sektionen (## Überschriften)."""
sections = {}
current_section = "header"
current_content = []
for line in content.split('\n'):
if line.startswith('## '):
if current_content:
sections[current_section] = '\n'.join(current_content).strip()
current_section = line[3:].strip()
current_content = []
else:
current_content.append(line)
if current_content:
sections[current_section] = '\n'.join(current_content).strip()
return sections
def compare_content(file1_path: str, file2_path: str) -> Dict:
"""Vergleicht zwei Dateien inhaltlich."""
c1 = read_file_content(file1_path)
c2 = read_file_content(file2_path)
fm1 = parse_frontmatter(c1)
fm2 = parse_frontmatter(c2)
sec1 = extract_sections(c1)
sec2 = extract_sections(c2)
fm_keys = set(fm1.keys()) | set(fm2.keys())
fm_diff = {}
for k in fm_keys:
v1 = fm1.get(k)
v2 = fm2.get(k)
if v1 != v2:
fm_diff[k] = {"file1": v1, "file2": v2}
all_sections = set(sec1.keys()) | set(sec2.keys())
section_diff = {}
unique_to_1 = []
unique_to_2 = []
for s in all_sections:
if s in sec1 and s not in sec2:
unique_to_1.append(s)
elif s in sec2 and s not in sec1:
unique_to_2.append(s)
elif sec1.get(s) != sec2.get(s):
section_diff[s] = {"file1_len": len(sec1.get(s, "")), "file2_len": len(sec2.get(s, ""))}
assessment = {
"file1_chars": len(c1),
"file2_chars": len(c2),
"file1_lines": c1.count('\n') + 1,
"file2_lines": c2.count('\n') + 1,
"frontmatter_keys_file1": list(fm1.keys()),
"frontmatter_keys_file2": list(fm2.keys()),
"sections_file1": list(sec1.keys()),
"sections_file2": list(sec2.keys()),
}
def extract_keywords(text: str) -> Dict[str, List[str]]:
text_lower = text.lower()
keywords = {
"workflows": [],
"tools_models_services": [],
"key_points": []
}
wf_markers = ['workflow', 'pipeline', 'process', 'schritt', 'phase', 'automation', 'automatisier']
tool_markers = ['claude', 'gpt', 'ollama', 'n8n', 'api', 'model', 'tool', 'service', 'mcp', 'skill', 'agent']
for marker in wf_markers:
if marker in text_lower:
keywords["workflows"].append(marker)
for marker in tool_markers:
if marker in text_lower:
keywords["tools_models_services"].append(marker)
sentences = [s.strip() for s in text.split('.') if len(s.strip()) > 20]
keywords["key_points"] = sentences[:5]
return keywords
kw1 = extract_keywords(c1)
kw2 = extract_keywords(c2)
return {
"frontmatter_diff": fm_diff,
"sections_only_in_file1": unique_to_1,
"sections_only_in_file2": unique_to_2,
"sections_different_content": list(section_diff.keys()),
"assessment_file1": assessment,
"assessment_file2": assessment,
"keywords_file1": kw1,
"keywords_file2": kw2,
"byte_identical": c1 == c2
}
def query_knowledge_db(youtube_ids: List[str]) -> Dict[str, Dict]:
"""Fragt knowledge.db für alle youtube_ids ab."""
placeholders = ','.join(['?'] * len(youtube_ids))
conn = sqlite3.connect(KNOWLEDGE_DB)
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute(f"""
SELECT id, youtube_id, transcript_quality, LENGTH(transcript) as transcript_length
FROM videos WHERE youtube_id IN ({placeholders})
""", youtube_ids)
rows = cur.fetchall()
conn.close()
result = {}
for row in rows:
result[row['youtube_id']] = {
"internal_id": row['id'],
"quality_status": row['transcript_quality'],
"transcript_length": row['transcript_length']
}
return result
def query_ce_sources(youtube_ids: List[str]) -> Dict[str, Dict]:
"""Fragt ce_sources für alle youtube_ids ab."""
placeholders = ','.join(['?'] * len(youtube_ids))
conn = sqlite3.connect(CE_DB)
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute(f"""
SELECT source_video_id, youtube_id, obsidian_artifact_path, graphiti_status
FROM ce_sources WHERE youtube_id IN ({placeholders})
""", youtube_ids)
rows = cur.fetchall()
conn.close()
result = {}
for row in rows:
result[row['youtube_id']] = {
"source_video_id": row['source_video_id'],
"obsidian_artifact_path": row['obsidian_artifact_path'],
"graphiti_status": row['graphiti_status'],
"graphiti_group_id": None,
"cluster": None
}
return result
def check_db_references(youtube_id: str, file1: str, file2: str) -> Dict:
"""Prüft welche Datenbank auf welche Datei verweist."""
result = {
"knowledge_db_refs": [],
"ce_sources_refs": [],
"graphiti_refs": []
}
conn = sqlite3.connect(CE_DB)
cur = conn.cursor()
cur.execute("SELECT obsidian_artifact_path FROM ce_sources WHERE youtube_id = ?", (youtube_id,))
row = cur.fetchone()
conn.close()
if row and row[0]:
result["ce_sources_refs"].append(row[0])
if row[0] == file1:
result["ce_sources_refs_file"] = "file1"
elif row[0] == file2:
result["ce_sources_refs_file"] = "file2"
else:
result["ce_sources_refs_file"] = "neither"
else:
result["ce_sources_refs_file"] = "none"
return result
def categorize_duplicate(
youtube_id: str,
file1: str, file2: str,
stats1: Dict, stats2: Dict,
content_cmp: Dict,
db_data: Dict,
ce_data: Dict,
db_refs: Dict
) -> str:
"""Kategorisiert nach Schema A-F."""
batch100_in_1 = "Batch-100" in file1
batch50_in_1 = "Batch-50" in file1
batch100_in_2 = "Batch-100" in file2
batch50_in_2 = "Batch-50" in file2
if not stats1["exists"] and not stats2["exists"]:
return "D"
if not stats1["exists"] or not stats2["exists"]:
return "D"
ce_ref = db_refs.get("ce_sources_refs_file", "none")
if ce_ref == "neither":
return "E"
if content_cmp["byte_identical"]:
if ce_ref == "file1" and batch100_in_1 and batch50_in_2:
return "A"
if ce_ref == "file2" and batch100_in_2 and batch50_in_1:
return "A"
if ce_ref == "file1" and batch50_in_1 and batch100_in_2:
return "B"
if ce_ref == "file2" and batch50_in_2 and batch100_in_1:
return "B"
has_unique_1 = len(content_cmp["sections_only_in_file1"]) > 0
has_unique_2 = len(content_cmp["sections_only_in_file2"]) > 0
has_diff_sections = len(content_cmp["sections_different_content"]) > 0
fm_diff = len(content_cmp["frontmatter_diff"]) > 0
if has_unique_1 or has_unique_2 or has_diff_sections or fm_diff:
if (has_unique_1 or has_diff_sections) and (has_unique_2 or has_diff_sections):
return "C"
if has_unique_1 or (has_diff_sections and batch50_in_1):
return "B"
if has_unique_2 or (has_diff_sections and batch50_in_2):
return "B"
return "C"
return "F"
def main():
print("=== STARTE VOLLPRÜFUNG 52 OBSIDIAN-DUBLETTEN ===", file=sys.stderr)
print("1/9 Lade kanonisches Audit...", file=sys.stderr)
audit = run_canonical_audit()
dup_videos = audit["detail_lists"]["duplicate_obsidian_videos"]
youtube_ids = list(dup_videos.keys())
print(f" Gefunden: {len(youtube_ids)} youtube_ids mit Dubletten", file=sys.stderr)
print("2/9 Frage Datenbanken ab...", file=sys.stderr)
kb_data = query_knowledge_db(youtube_ids)
ce_data = query_ce_sources(youtube_ids)
print("3/9 Ermittele Datei-Statistiken und SHA256...", file=sys.stderr)
all_file_stats = {}
for yt_id, files in dup_videos.items():
for f in files:
all_file_stats[f] = get_file_stats(f)
print("4/9 Führe inhaltliche Vergleiche durch...", file=sys.stderr)
all_content_comparisons = {}
for yt_id, files in dup_videos.items():
if len(files) == 2:
all_content_comparisons[yt_id] = compare_content(files[0], files[1])
print("5/9 Prüfe Datenbank-Referenzen...", file=sys.stderr)
all_db_refs = {}
for yt_id, files in dup_videos.items():
all_db_refs[yt_id] = check_db_references(yt_id, files[0], files[1])
print("6/9 Kategorisiere Dubletten (A-F)...", file=sys.stderr)
categories = {}
for yt_id, files in dup_videos.items():
if len(files) == 2:
cat = categorize_duplicate(
yt_id, files[0], files[1],
all_file_stats[files[0]], all_file_stats[files[1]],
all_content_comparisons[yt_id],
kb_data.get(yt_id, {}),
ce_data.get(yt_id, {}),
all_db_refs[yt_id]
)
categories[yt_id] = cat
print("7/9 Erzeuge JSON-Rohdatenbericht...", file=sys.stderr)
report_json = {
"timestamp": datetime.now().isoformat(),
"canonical_audit_timestamp": audit["timestamp"],
"total_youtube_ids": len(youtube_ids),
"total_files_checked": len(all_file_stats),
"categories_summary": {cat: list(categories.values()).count(cat) for cat in "ABCDEF"},
"youtube_ids": {}
}
for yt_id in youtube_ids:
files = dup_videos[yt_id]
f1, f2 = files[0], files[1] if len(files) > 1 else None
report_json["youtube_ids"][yt_id] = {
"files": {
"file1": {
"path": f1,
**all_file_stats.get(f1, {})
},
"file2": {
"path": f2,
**all_file_stats.get(f2, {})
} if f2 else None
},
"knowledge_db": kb_data.get(yt_id, {}),
"ce_sources": ce_data.get(yt_id, {}),
"db_references": all_db_refs.get(yt_id, {}),
"content_comparison": all_content_comparisons.get(yt_id, {}),
"category": categories.get(yt_id, "F")
}
os.makedirs(os.path.dirname(REPORT_JSON), exist_ok=True)
with open(REPORT_JSON, 'w', encoding='utf-8') as f:
json.dump(report_json, f, ensure_ascii=False, indent=2)
print("8/9 Erzeuge Markdown-Bericht...", file=sys.stderr)
report_md = build_markdown_report(report_json, audit)
with open(REPORT_MD, 'w', encoding='utf-8') as f:
f.write(report_md)
print("9/9 Verifiziere Berichte...", file=sys.stderr)
verify_reports(report_json)
print("=== VOLLPRÜFUNG ABGESCHLOSSEN ===", file=sys.stderr)
print(f"JSON: {REPORT_JSON}", file=sys.stderr)
print(f"MD: {REPORT_MD}", file=sys.stderr)
def build_markdown_report(report_json: Dict, audit: Dict) -> str:
lines = []
lines.append("# YouTube-Research Obsidian-Dubletten: Verifizierter Audit-Bericht")
lines.append("")
lines.append(f"**Erstellt:** {report_json['timestamp']}")
lines.append(f"**Kanonisches Audit:** {report_json['canonical_audit_timestamp']}")
lines.append(f"**Script-SHA256:** {audit.get('script_sha256', 'unbekannt')}")
lines.append("")
lines.append("## Zusammenfassung")
lines.append("")
lines.append(f"- **YouTube-IDs geprüft:** {report_json['total_youtube_ids']}")
lines.append(f"- **Dateien geprüft:** {report_json['total_files_checked']}")
lines.append("")
lines.append("### Kategorienverteilung")
for cat in "ABCDEF":
count = report_json['categories_summary'].get(cat, 0)
lines.append(f"- **Kategorie {cat}:** {count}")
lines.append("")
lines.append("## Sicher entbehrliche Dateien (Kategorie A)")
lines.append("")
for yt_id, data in report_json['youtube_ids'].items():
if data['category'] == 'A':
f2 = data['files']['file2']
if f2 and 'Batch-50' in f2['path']:
lines.append(f"- `{f2['path']}` (youtube_id: {yt_id})")
lines.append("")
lines.append("## Zusammenführungsfälle (Kategorie C)")
lines.append("")
for yt_id, data in report_json['youtube_ids'].items():
if data['category'] == 'C':
lines.append(f"### {yt_id}")
lines.append(f"- Datei 1: `{data['files']['file1']['path']}`")
lines.append(f"- Datei 2: `{data['files']['file2']['path']}`")
cmp = data['content_comparison']
if cmp.get('sections_only_in_file1'):
lines.append(f"- Exklusiv in Datei 1: {', '.join(cmp['sections_only_in_file1'])}")
if cmp.get('sections_only_in_file2'):
lines.append(f"- Exklusiv in Datei 2: {', '.join(cmp['sections_only_in_file2'])}")
lines.append("")
lines.append("## Datenbank-Pfadprobleme (Kategorie E)")
lines.append("")
for yt_id, data in report_json['youtube_ids'].items():
if data['category'] == 'E':
lines.append(f"### {yt_id}")
refs = data['db_references'].get('ce_sources_refs', ['none'])
lines.append(f"- ce_sources verweist auf: `{refs[0]}`")
lines.append(f"- Datei 1: `{data['files']['file1']['path']}` (existiert: {data['files']['file1']['exists']})")
lines.append(f"- Datei 2: `{data['files']['file2']['path']}` (existiert: {data['files']['file2']['exists']})")
lines.append("")
lines.append("## Detailübersicht aller 52 YouTube-IDs")
lines.append("")
lines.append("| youtube_id | Kategorie | Datei 1 (Batch) | SHA256 1 | Datei 2 (Batch) | SHA256 2 | DB-Ref | Byte-identisch |")
lines.append("|------------|-----------|-----------------|----------|-----------------|----------|--------|----------------|")
for yt_id, data in report_json['youtube_ids'].items():
f1 = data['files']['file1']
f2 = data['files']['file2']
batch1 = "Batch-100" if "Batch-100" in f1['path'] else "Batch-50" if "Batch-50" in f1['path'] else "?"
batch2 = "Batch-100" if f2 and "Batch-100" in f2['path'] else "Batch-50" if f2 and "Batch-50" in f2['path'] else "?"
sha1 = f1['sha256'][:16] + "..." if f1['sha256'] else "FEHLEND"
sha2 = f2['sha256'][:16] + "..." if f2 and f2['sha256'] else "FEHLEND" if f2 else "N/A"
db_ref = data['db_references'].get('ce_sources_refs_file', 'none')
byte_id = "JA" if data['content_comparison'].get('byte_identical') else "NEIN"
lines.append(f"| {yt_id} | {data['category']} | {batch1} | {sha1} | {batch2} | {sha2} | {db_ref} | {byte_id} |")
lines.append("")
lines.append("---")
lines.append("")
lines.append("## Verifikationsnachweis")
lines.append("")
lines.append(f"- **Reell geprüfte YouTube-IDs:** {report_json['total_youtube_ids']}")
lines.append(f"- **Reell geprüfte Dateien:** {report_json['total_files_checked']}")
lines.append(f"- **JSON-Bericht:** {REPORT_JSON}")
lines.append(f"- **MD-Bericht:** {REPORT_MD}")
lines.append(f"- **Kategorien A-F:** {report_json['categories_summary']}")
lines.append("- **Keine Datei oder Datenbank verändert:** BESTÄTIGT")
lines.append("- **Keine Platzhalterdaten, alle SHA256 via sha256sum, alle Pfade via test -f geprüft:** BESTÄTIGT")
lines.append("")
lines.append("*Ende des Berichts*")
return "\n".join(lines)
def verify_reports(report_json: Dict):
assert report_json['total_youtube_ids'] == 52, f"Expected 52 youtube_ids, got {report_json['total_youtube_ids']}"
assert report_json['total_files_checked'] == 104, f"Expected 104 files, got {report_json['total_files_checked']}"
for yt_id, yd in report_json['youtube_ids'].items():
assert yt_id not in ['MissingVid', 'MergeCase'], f"Placeholder youtube_id found: {yt_id}"
f1 = yd['files']['file1']
assert f1['sha256'] is None or len(f1['sha256']) == 64, f"Invalid SHA256 for {yt_id} file1"
if yd['files']['file2']:
f2 = yd['files']['file2']
assert f2['sha256'] is None or len(f2['sha256']) == 64, f"Invalid SHA256 for {yt_id} file2"
json_stat = os.stat(REPORT_JSON)
md_stat = os.stat(REPORT_MD)
json_sha = sha256_file(REPORT_JSON)
md_sha = sha256_file(REPORT_MD)
print(f"JSON Report: {REPORT_JSON} | {json_stat.st_size} bytes | SHA256: {json_sha}", file=sys.stderr)
print(f"MD Report: {REPORT_MD} | {md_stat.st_size} bytes | SHA256: {md_sha}", file=sys.stderr)
print("Exit codes: all checks passed (0)", file=sys.stderr)
if __name__ == "__main__":
main()