Spaces:
Sleeping
Sleeping
File size: 6,608 Bytes
b9f4b13 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | """
SQLite Evidence Cache — Persists mined raw evidence with TTL to prevent redundant scraping.
"""
from __future__ import annotations
import json
import logging
import sqlite3
import time
from pathlib import Path
from typing import Sequence
from src.mining.provider import RawEvidence
from src.models.common import DataSource
logger = logging.getLogger(__name__)
DEFAULT_CACHE_DB_PATH = ".cache/ventureforge.db"
DEFAULT_CACHE_TTL_SECONDS = 86400.0 # 24 hours
class SQLiteEvidenceCache:
"""SQLite-backed cache for RawEvidence keyed by domain with TTL expiration."""
def __init__(self, db_path: str = DEFAULT_CACHE_DB_PATH) -> None:
self.db_path = db_path
self._init_db()
def _get_connection(self) -> sqlite3.Connection:
db_file = Path(self.db_path)
db_file.parent.mkdir(parents=True, exist_ok=True)
return sqlite3.connect(str(db_file), check_same_thread=False)
def _init_db(self) -> None:
try:
with self._get_connection() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS mined_evidence_cache (
domain TEXT NOT NULL,
source TEXT NOT NULL,
raw_json TEXT NOT NULL,
mined_at REAL NOT NULL,
PRIMARY KEY (domain, source)
)
"""
)
conn.commit()
except Exception as e:
logger.warning(f"[SQLiteEvidenceCache] Failed to initialize cache table: {e}")
def get(
self,
domain: str,
max_age_seconds: float = DEFAULT_CACHE_TTL_SECONDS,
) -> list[RawEvidence] | None:
"""Retrieve cached evidence for a domain if fresher than max_age_seconds."""
domain_key = domain.strip().lower()
now = time.time()
try:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"""
SELECT raw_json, mined_at FROM mined_evidence_cache
WHERE domain = ?
""",
(domain_key,),
)
rows = cursor.fetchall()
if not rows:
return None
results: list[RawEvidence] = []
for raw_json, mined_at in rows:
if now - mined_at > max_age_seconds:
logger.info(f"[SQLiteEvidenceCache] Cache expired for domain '{domain}'.")
return None
items_data = json.loads(raw_json)
for item in items_data:
# Never serve synthetic/fabricated evidence from cache
if item.get("metadata", {}).get("synthetic"):
continue
source_val = item.get("source", "web")
try:
source_enum = DataSource(source_val)
except ValueError:
source_enum = DataSource.WEB
results.append(
RawEvidence(
text=item.get("text", ""),
url=item.get("url", ""),
source=source_enum,
title=item.get("title", ""),
author=item.get("author", ""),
score=item.get("score", 0),
metadata=item.get("metadata", {}),
)
)
if results:
logger.info(
f"[SQLiteEvidenceCache] Cache HIT for domain='{domain}' "
f"({len(results)} items loaded from SQLite)."
)
return results
except Exception as e:
logger.warning(f"[SQLiteEvidenceCache] Error reading cache for domain '{domain}': {e}")
return None
def set(self, domain: str, evidence: Sequence[RawEvidence]) -> None:
"""Save evidence items grouped by source for a domain with current timestamp."""
if not evidence:
return
domain_key = domain.strip().lower()
now = time.time()
# Group by source
by_source: dict[str, list[dict]] = {}
for item in evidence:
# Never cache synthetic/fabricated evidence
if item.metadata.get("synthetic"):
continue
src_name = item.source.value if hasattr(item.source, "value") else str(item.source)
item_dict = {
"text": item.text,
"url": item.url,
"source": src_name,
"title": item.title,
"author": item.author,
"score": item.score,
"metadata": item.metadata,
}
by_source.setdefault(src_name, []).append(item_dict)
try:
with self._get_connection() as conn:
cursor = conn.cursor()
for src_name, items_list in by_source.items():
cursor.execute(
"""
INSERT OR REPLACE INTO mined_evidence_cache (domain, source, raw_json, mined_at)
VALUES (?, ?, ?, ?)
""",
(domain_key, src_name, json.dumps(items_list), now),
)
conn.commit()
logger.info(
f"[SQLiteEvidenceCache] Saved {len(evidence)} evidence items for domain='{domain}'."
)
except Exception as e:
logger.warning(f"[SQLiteEvidenceCache] Error writing cache for domain '{domain}': {e}")
def clear(self, domain: str | None = None) -> None:
"""Clear cache entries for a specific domain or entire cache if domain is None."""
try:
with self._get_connection() as conn:
if domain is None:
conn.execute("DELETE FROM mined_evidence_cache")
else:
conn.execute(
"DELETE FROM mined_evidence_cache WHERE domain = ?",
(domain.strip().lower(),),
)
conn.commit()
except Exception as e:
logger.warning(f"[SQLiteEvidenceCache] Error clearing cache: {e}")
|