#!/usr/bin/env python3
import sqlite3
import os
import subprocess
from pathlib import Path
def find_databases():
"""Find all SQLite databases in /opt/struktur and /var/lib/sma-data"""
db_paths = []
search_paths = [
"/opt/struktur",
"/var/lib/sma-data"
]
for search_path in search_paths:
if os.path.exists(search_path):
for root, dirs, files in os.walk(search_path):
# Skip backup directories to avoid too much noise
dirs[:] = [d for d in dirs if not d.startswith('.') and 'backup' not in d.lower()]
for file in files:
if file.endswith('.db'):
full_path = os.path.join(root, file)
db_paths.append(full_path)
return db_paths
def inspect_database(db_path):
"""Inspect a single database and return schema info"""
result = {
'path': db_path,
'exists': os.path.exists(db_path),
'readable': False,
'integrity_ok': False,
'foreign_key_issues': [],
'tables': {},
'error': None
}
if not result['exists']:
result['error'] = "Datei existiert nicht"
return result
try:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
result['readable'] = True
# Integrity check
cursor.execute("PRAGMA integrity_check;")
integrity_result = cursor.fetchone()
result['integrity_ok'] = integrity_result[0] == 'ok' if integrity_result else False
# Foreign key check
cursor.execute("PRAGMA foreign_key_check;")
fk_issues = cursor.fetchall()
result['foreign_key_issues'] = [dict(row) for row in fk_issues]
# Get tables
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name;")
tables = cursor.fetchall()
for table in tables:
table_name = table[0]
# Skip sqlite_sequence
if table_name == 'sqlite_sequence':
continue
# Get table info
cursor.execute(f"PRAGMA table_info({table_name});")
columns = [dict(row) for row in cursor.fetchall()]
# Get foreign keys
cursor.execute(f"PRAGMA foreign_key_list({table_name});")
foreign_keys = [dict(row) for row in cursor.fetchall()]
# Get indexes
cursor.execute(f"PRAGMA index_list({table_name});")
indexes = [dict(row) for row in cursor.fetchall()]
# Get row count
try:
cursor.execute(f"SELECT COUNT(*) FROM {table_name};")
row_count = cursor.fetchone()[0]
except:
row_count = -1 # Error counting
result['tables'][table_name] = {
'columns': columns,
'foreign_keys': foreign_keys,
'indexes': indexes,
'row_count': row_count
}
conn.close()
except Exception as e:
result['error'] = str(e)
return result
def find_direct_sql_access():
"""Find direct SQL accesses in the codebase"""
sql_patterns = [
r'sqlite3\.connect',
r'\.execute\(',
r'\.executescript\(',
r'conn\s*=',
r'cursor\s*=',
]
results = []
exclude_dirs = ['.git', '__pycache__', 'backups', 'reports', 'node_modules', '.venv', 'venv']
for root, dirs, files in os.walk('/opt/struktur'):
# Modify dirs in-place to exclude directories
dirs[:] = [d for d in dirs if d not in exclude_dirs and not d.startswith('.')]
for file in files:
if file.endswith('.py'):
filepath = os.path.join(root, file)
try:
with open(filepath, 'r', encoding='utf-8') as f:
lines = f.readlines()
for i, line in enumerate(lines, 1):
line_stripped = line.strip()
# Skip comments and empty lines
if line_stripped.startswith('#') or not line_stripped:
continue
# Check for SQL patterns
for pattern in sql_patterns:
if pattern in line:
# Get context (previous and next line)
context_before = lines[max(0, i-2)-1:i-1] if i > 1 else []
context_after = lines[i:min(len(lines), i+2)] if i < len(lines) else []
results.append({
'file': filepath,
'line': i,
'code': line.rstrip(),
'context_before': [ctx.rstrip() for ctx in context_before],
'context_after': [ctx.rstrip() for ctx in context_after]
})
break # Only count once per line
except Exception as e:
# Skip files we can't read
pass
return results
def get_database_paths_from_code():
"""Extract database path configurations from code"""
path_vars = ['RADAR_DB', 'CASES_DB', 'POSTS_DB', 'SMA_DATABASE_PATH']
results = {}
for root, dirs, files in os.walk('/opt/struktur'):
# Exclude backup and report directories
dirs[:] = [d for d in dirs if d not in ['.git', '__pycache__', 'backups', 'reports', 'node_modules', '.venv', 'venv'] and not d.startswith('.')]
for file in files:
if file.endswith('.py'):
filepath = os.path.join(root, file)
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
# Look for path assignments
for var in path_vars:
# Pattern for Path(...) or direct string assignments
import re
patterns = [
rf'{var}\s*=\s*Path\(["\']([^"\']*)["\']\)',
rf'{var}\s*=\s*["\']([^"\']*)["\']',
rf'{var}\s*=\s*os\.path\.join\([^)]*\)',
rf'{var}\s*=\s*Path\([^)]*\)',
]
for pattern in patterns:
matches = re.findall(pattern, content)
if matches:
if var not in results:
results[var] = []
results[var].extend(matches)
except:
pass
return results
def check_backup_locations():
"""Check for known backup locations"""
backup_locations = [
"/root/backups-h012/",
"/root/backups-nginx/",
"/opt/struktur/backups/",
"/opt/struktur/reports/",
"/home/hermes/AA-*.md"
]
results = {}
for loc in backup_locations:
if '*' in loc:
# Handle glob patterns
import glob
matches = glob.glob(loc)
results[loc] = matches
else:
if os.path.exists(loc):
if os.path.isdir(loc):
try:
files = os.listdir(loc)
results[loc] = f"Directory with {len(files)} items"
except:
results[loc] = "Directory (access error)"
else:
results[loc] = "File exists"
else:
results[loc] = "Not found"
return results
def main():
print("=" * 80)
print("ARGUS - DATEN- UND PERSISTENZKONTROLLE FÜR AA-053")
print("=" * 80)
print()
# 1. Alle Datenbanken finden
print("1. SUCHE NACH SQLITE-DATENBANKEN")
print("-" * 40)
db_paths = find_databases()
print(f"Gefundene Datenbanken ({len(db_paths)}):")
for db in sorted(db_paths):
print(f" - {db}")
print()
# 2. Jede Datenbank inspizieren
print("2. DATENBANK-INSPEKTION")
print("-" * 40)
db_inspections = []
for db_path in sorted(db_paths):
print(f"\nInspektion: {db_path}")
inspection = inspect_database(db_path)
db_inspections.append(inspection)
if inspection['error']:
print(f" FEHLER: {inspection['error']}")
else:
print(f" Lesbar: {'Ja' if inspection['readable'] else 'Nein'}")
print(f" Integrity OK: {'Ja' if inspection['integrity_ok'] else 'Nein'}")
print(f" FK-Issues: {len(inspection['foreign_key_issues'])}")
print(f" Tabellen: {len(inspection['tables'])}")
# Show tables with row counts
for table_name, table_info in inspection['tables'].items():
print(f" - {table_name}: {table_info['row_count']} Zeilen, {len(table_info['columns'])} Spalten")
# Show any FK issues
if inspection['foreign_key_issues']:
print(f" FK-Probleme:")
for fk in inspection['foreign_key_issues'][:3]: # Show first 3
print(f" Tabelle {fk['table']}: {fk['foreign_key']} -> {fk['parent_table']}({fk['parent_key']})")
print()
# 3. Direkter SQL-Zugriff im Code
print("3. SUCHE NACH DIREKTEN SQL-ZUGRIFFEN IM CODE")
print("-" * 40)
direct_sql = find_direct_sql_access()
print(f"Gefundene direkte SQL-Zugriffe: {len(direct_sql)}")
# Group by file to avoid too much output
sql_by_file = {}
for access in direct_sql:
filepath = access['file']
if filepath not in sql_by_file:
sql_by_file[filepath] = []
sql_by_file[filepath].append(access)
# Show files with most SQL accesses first
sorted_files = sorted(sql_by_file.items(), key=lambda x: len(x[1]), reverse=True)
for filepath, accesses in sorted_files[:10]: # Top 10 files
print(f"\n{filepath} ({len(accesses)} Zugriffe):")
for access in accesses[:3]: # Show first 3 accesses per file
print(f" Zeile {access['line']}: {access['code']}")
if access['context_before']:
print(f" Vorher: {access['context_before'][-1] if access['context_before'] else ''}")
if access['context_after']:
print(f" Nachher: {access['context_after'][0] if access['context_after'] else ''}")
if len(sorted_files) > 10:
print(f"\n... und {len(sorted_files) - 10} weitere Dateien mit SQL-Zugriffen")
print()
# 4. Datenbankpfade aus Code
print("4. DATENBANKPfade AUS CODE")
print("-" * 40)
path_configs = get_database_paths_from_code()
for var, paths in path_configs.items():
if paths:
print(f"{var}:")
for path in set(paths): # Deduplicate
print(f" - {path}")
print()
# 5. Backup-Locations prüfen
print("5. BACKUP-/ROLLBACK-NACHWEISE")
print("-" * 40)
backups = check_backup_locations()
for location, info in backups.items():
print(f"{location}: {info}")
print()
# 6. MAS/AS-Isolation prüfen (basierend auf bekannten Pfaden)
print("6. MAS/AS-ISOLATIONSKONTROLLE")
print("-" * 40)
# Known brand-specific paths from documentation
brand_paths = {
'Mallorca AirServices (MAS)': [
'/var/lib/sma-data/signals.db',
'/var/lib/sma-data/cases.db',
'/opt/struktur/social-media-radar/signals/radar.db'
],
'Agent Solutions (AS)': [
'/opt/struktur/social-media-agent/db/posts.db',
'/opt/struktur/social-media-agent/cases.db'
]
}
for brand, paths in brand_paths.items():
print(f"{brand}:")
for path in paths:
exists = os.path.exists(path)
print(f" - {path}: {'EXISTIERT' if exists else 'FEHLT'}")
print()
# Summary of findings
print("=" * 80)
print("ZUSAMMENFASSUNG DER BEFRAGE")
print("=" * 80)
# Count total databases with issues
problematic_dbs = [db for db in db_inspections if not db['integrity_ok'] or db['error'] or db['foreign_key_issues']]
if problematic_dbs:
print(f"⚠️ PROBLEMATISCHE DATENBANKEN: {len(problematic_dbs)} von {len(db_inspections)}")
for db in problematic_dbs:
issues = []
if db['error']:
issues.append(f"Fehler: {db['error']}")
if not db['integrity_ok']:
issues.append("Integrity-Check fehlgeschlagen")
if db['foreign_key_issues']:
issues.append(f"{len(db['foreign_key_issues'])} FK-Verstöße")
print(f" - {db['path']}: {', '.join(issues)}")
else:
print("✅ Alle Datenbanken zeigen keine Integritätsprobleme")
# Direct SQL access concerns
if len(direct_sql) > 50: # Arbitrary threshold
print(f"⚠️ HOHE ANZAHL DIREKTER SQL-ZUGRIFFE: {len(direct_sql)} Stellen im Code")
print(" Dies könnte auf fehlende Abstraktion über Repository/Service-Schichten hindeuten")
else:
print(f"✅ Direkte SQL-Zugriffe im akzeptablen Bereich: {len(direct_sql)} Stellen")
# Check for overlap between brands
mas_paths = set(brand_paths['Mallorca AirServices (MAS)'])
as_paths = set(brand_paths['Agent Solutions (AS)'])
overlap = mas_paths & as_paths
if overlap:
print(f"⚠️ ÜBERLAPPENDE DATENBANKPfade ZWISCHEN MARKEN: {overlap}")
else:
print("✅ Keine offensichtlichen Überschneidungen in den bekannten Marken-Datenbankpfaden")
print()
print("HINWEIS: Dies ist eine schreibgeschützte Analyse. Keine produktiven Änderungen wurden vorgenommen.")
print("Für detaillierte Untersuchung einzelner Befunde können gezielte Follow-up-Anfragen gestellt werden.")
if __name__ == "__main__":
main()