#!/usr/bin/env python3
"""
scope_policy_checker.py
Wiederverwendbares Modul fuer alle Graphiti-Import-Skripte.
Laedt import_scope_policy.yaml und prueft Dateipfade und Content
gegen Allowlist, Denylist und Content-Filter.
Verwendung:
from scope_policy_checker import ScopeChecker
checker = ScopeChecker()
ok, reason = checker.check_path("/opt/obsidian-vault/YouTube-Research/Welle-1/ID-034.md")
ok, reason = checker.check_content(content_string)
clean_content = checker.clean_content(content_string)
"""
import re
import os
from pathlib import Path
POLICY_PATH = "/opt/struktur/graphiti/import_scope_policy.yaml"
# Private Labels und Video-Meta direkt als Konstanten (kein yaml-Import noetig)
PRIVATE_TERMS = [
"LüftungsProfi", "Lueftungsprofi", "LP-spezifisch",
"LP-Skills", "LP-spezifische", "Agent Solutions",
"projekte-LP", "projekte-AG", "inkasso2025",
]
DENIED_PATH_PATTERNS = [
"LP-CONTEXT", "LP-SPECIFIC", "LP-",
"AG-CONTEXT", "AG-SPECIFIC",
"03-Projekte", "06-Daily-Notes", "00-Staging", "07-Archiv",
"Use-Cases/Lueftungsprofi", "Use-Cases/LP-",
"Lueftungsprofi", "LUEFTUNGSPROFI",
".env", "credentials", "secrets",
"SYNC-LP", "SYNC-AG",
"Volltranskript", "Rohdaten",
]
DENIED_DIR_NAMES = {
"LP-CONTEXT", "LP-SPECIFIC", "AG-CONTEXT", "AG-SPECIFIC",
"03-Projekte", "06-Daily-Notes", "07-Archiv", "00-Staging",
".obsidian", ".claude", ".claudian", "08-Attachments",
"INSTALL", "sessions",
}
ALLOWED_ROOTS = [
"/opt/obsidian-vault/YouTube-Research/Welle-1",
"/opt/obsidian-vault/YouTube-Research/Welle-2",
"/opt/obsidian-vault/YouTube-Research/Pilot",
]
VIDEO_META_PATTERNS = [
re.compile(r"(?i)(the|das|this|dieses?) video[^.]{0,80}?(discusses|is about|covers|erklaert|zeigt|handelt|bespricht)[^.]*\."),
re.compile(r"(?i)(is|are) (hosted|published|available|featured) on youtube[^.]*\."),
re.compile(r"(?i)(published|posted|created|produced|released).{0,30}(youtube|video)[^.]*\."),
re.compile(r"(?i)(channel|kanal)[^.]{0,40}provides (content|tutorials|insights|videos)[^.]*\."),
re.compile(r"(?i)the episode[^.]{0,60}?(is about|discusses|titled)[^.]*\."),
re.compile(r"(?i)(Webronaq|Jack Roberts|Der KI-Doktor|Mike Mildenberger|Julian Ivanov|Marc De Fanti|Philip Thomas|AI mit Arnie|Alex Sprogis|Prompt Engineer)[^.]{0,40}(published|posted|created|produced|made|released)[^.]*\."),
]
TAUTOLOGY_PATTERNS = [
re.compile(r"(?i)\w+.s\s+creator\s+created\s+\w+"),
re.compile(r"(?i)\w+\s+is\s+used\s+in\s+\w+\s+environment"),
re.compile(r"(?i)^\s*-\s+the video (about|on|regarding) \w+ is hosted on youtube"),
]
PRIVATE_TERM_RE = [re.compile(re.escape(t)) for t in PRIVATE_TERMS]
class ScopeChecker:
"""Prueft Pfade und Content gegen die Import-Scope-Policy."""
def check_path(self, filepath: str) -> tuple:
"""
Gibt (True, "ok") oder (False, "Grund") zurueck.
Denylist gewinnt immer vor Allowlist.
"""
fp = str(filepath)
path = Path(fp)
# 1. Denylist: Directory-Names in Pfad pruefen
parts = path.parts
for part in parts:
if part in DENIED_DIR_NAMES:
return False, f"DENY: Verzeichnis '{part}' ist gesperrt"
# 2. Denylist: Pattern in absolutem Pfad
for pattern in DENIED_PATH_PATTERNS:
if pattern in fp:
return False, f"DENY: Pfadpattern '{pattern}' ist gesperrt"
# 3. Allowlist: Muss unter einem erlaubten Root liegen
allowed = any(fp.startswith(root) for root in ALLOWED_ROOTS)
if not allowed:
return False, f"DENY: Pfad liegt nicht unter einem erlaubten Root (allowed_roots in Policy)"
return True, "ok"
def check_content(self, content: str) -> tuple:
"""
Prueft ob Content private Terme enthaelt.
Gibt (True, "ok") oder (False, "Grund") zurueck.
"""
for pat in PRIVATE_TERM_RE:
m = pat.search(content)
if m:
return False, f"DENY: Private Term '{m.group()}' im Content"
return True, "ok"
def check_frontmatter(self, frontmatter: dict) -> tuple:
"""
Prueft YAML-Frontmatter auf Pflichtfelder fuer YouTube-Research.
"""
gr = str(frontmatter.get("graphiti_ready", "false")).lower()
if gr not in ("true", "yes", "1"):
return False, f"DENY: graphiti_ready={gr} (muss true sein)"
iid = str(frontmatter.get("internal_id", ""))
if iid in {"046"}:
return False, f"DENY: internal_id={iid} ist explizit gesperrt"
return True, "ok"
def clean_content(self, content: str) -> str:
"""
Bereinigt Content: entfernt Video-Meta-Saetze, Tautologien,
private Label-Vorkommen zeilenweise.
"""
lines = content.split("\n")
cleaned = []
for line in lines:
stripped = line.strip()
# Leerzeilen und Sektionsheader immer behalten
if not stripped or stripped.startswith("#"):
cleaned.append(line)
continue
# Private Terme: ganze Zeile entfernen
has_private = any(pat.search(stripped) for pat in PRIVATE_TERM_RE)
if has_private:
continue
# Video-Meta-Saetze: ganze Zeile entfernen
has_video_meta = any(pat.search(stripped) for pat in VIDEO_META_PATTERNS)
if has_video_meta:
continue
# Tautologien: ganze Zeile entfernen
has_tautology = any(pat.search(stripped) for pat in TAUTOLOGY_PATTERNS)
if has_tautology:
continue
cleaned.append(line)
return "\n".join(cleaned)
def strip_blocked_sections(self, content: str,
blocked=("Volltext-Kontext", "Quellenanker")) -> str:
"""
Entfernt vollstaendige Markdown-Sektionen aus dem Content.
Gibt bereinigten Content zurueck.
"""
lines = content.split("\n")
result = []
skip = False
for line in lines:
if line.startswith("## "):
sec_name = line[3:].strip()
skip = any(b in sec_name for b in blocked)
if not skip:
result.append(line)
return "\n".join(result)
def neutral_source_line(self, meta: dict) -> str:
"""Erzeugt neutrale Quellenzeile ohne private Labels."""
title = meta.get("title", "").strip()
channel = meta.get("channel", "").strip()
yt_id = meta.get("youtube_id", "").strip()
lang = meta.get("language", "").strip()
# Sicherheitsfilter: keine privaten Terme in Metadaten
for term in PRIVATE_TERMS:
title = title.replace(term, "").strip()
channel = channel.replace(term, "").strip()
return f"Quelle: {title} | Kanal: {channel} | ID: {yt_id} | Sprache: {lang}"
# Singleton fuer direkten Import
_checker = None
def get_checker() -> ScopeChecker:
global _checker
if _checker is None:
_checker = ScopeChecker()
return _checker