mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
411 lines
18 KiB
Python
411 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Smart Ingest Watcher (Daemon)
|
|
=============================
|
|
Monitors watched directories for new PDFs. On detection:
|
|
1. Extract arXiv ID / DOI from filename + PDF header
|
|
2. Fetch metadata from arXiv API if arXiv ID found
|
|
3. Ingest into Zotero with proper metadata
|
|
4. Move to the canonical archive directory
|
|
5. Log everything to the index
|
|
|
|
Usage:
|
|
# Run once (foreground, dry-run)
|
|
python3 ingest_watcher.py --once --dry-run
|
|
|
|
# Run as daemon
|
|
python3 ingest_watcher.py --daemon
|
|
|
|
# Add a watch directory
|
|
python3 ingest_watcher.py --watch ~/Downloads --daemon
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Set
|
|
|
|
# ── Paths ─────────────────────────────────────────────────────────────────────
|
|
ZOTERO_DB = Path.home() / "Zotero" / "zotero.sqlite"
|
|
INDEX_DB = Path.home() / "Research Stack" / "data" / "substrate_index.db"
|
|
CANONICAL_DIR = Path.home() / "Downloads" / "data" / "Downloads_from_internet" / "Deep Research" / "alphaXiv_PDFs_2026_04"
|
|
STATE_FILE = Path.home() / "Research Stack" / "data" / "ingest_watcher_state.json"
|
|
LOG_FILE = Path.home() / "Research Stack" / "data" / "ingest_watcher.log"
|
|
|
|
CANONICAL_DIR.mkdir(parents=True, exist_ok=True)
|
|
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# ── Config ────────────────────────────────────────────────────────────────────
|
|
DEFAULT_WATCH_DIRS = [
|
|
str(Path.home() / "Downloads"),
|
|
str(Path.home() / "Downloads" / "data" / "Downloads_from_internet"),
|
|
]
|
|
CHECK_INTERVAL = 30 # seconds
|
|
|
|
# ── Data Classes ─────────────────────────────────────────────────────────────
|
|
@dataclass
|
|
class DiscoveredPDF:
|
|
path: Path
|
|
arxiv_id: Optional[str] = None
|
|
doi: Optional[str] = None
|
|
title_guess: str = ""
|
|
hash: str = ""
|
|
|
|
# ── arXiv Client (lightweight) ────────────────────────────────────────────────
|
|
class ArxivClient:
|
|
BASE_QUERY = "https://export.arxiv.org/api/query"
|
|
BASE_PDF = "https://arxiv.org/pdf"
|
|
|
|
def fetch_meta(self, arxiv_id: str) -> Optional[Dict]:
|
|
clean = re.sub(r"v\d+$", "", arxiv_id)
|
|
url = f"{self.BASE_QUERY}?id_list={clean}&max_results=1"
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "ScienceHub-Watcher/0.1"})
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
xml = resp.read().decode("utf-8")
|
|
return self._parse(xml, clean)
|
|
except Exception as e:
|
|
self._log(f"[arxiv] Failed to fetch {arxiv_id}: {e}")
|
|
return None
|
|
|
|
def _parse(self, xml: str, arxiv_id: str) -> Optional[Dict]:
|
|
import xml.etree.ElementTree as ET
|
|
ns = {"atom": "http://www.w3.org/2005/Atom", "arxiv": "http://arxiv.org/schemas/atom"}
|
|
root = ET.fromstring(xml)
|
|
entry = root.find("atom:entry", ns)
|
|
if entry is None:
|
|
return None
|
|
def tag(t):
|
|
el = entry.find(f"atom:{t}", ns)
|
|
return el.text.strip() if el is not None else ""
|
|
authors = [a.find("atom:name", ns).text for a in entry.findall("atom:author", ns)]
|
|
cat_el = entry.find("arxiv:primary_category", ns)
|
|
primary = cat_el.attrib.get("term", "") if cat_el is not None else ""
|
|
cats = [c.attrib.get("term", "") for c in entry.findall("atom:category", ns)]
|
|
return {
|
|
"arxiv_id": arxiv_id,
|
|
"title": tag("title").replace("\n", " "),
|
|
"authors": authors,
|
|
"summary": tag("summary").replace("\n", " "),
|
|
"published": tag("published")[:4],
|
|
"updated": tag("updated"),
|
|
"primary_category": primary,
|
|
"categories": cats,
|
|
"pdf_url": f"{self.BASE_PDF}/{arxiv_id}.pdf",
|
|
}
|
|
|
|
def _log(self, msg: str):
|
|
print(msg, file=sys.stderr)
|
|
|
|
# ── Zotero Writer (same as sciencehub_mcp but minimal) ──────────────────────
|
|
class ZoteroWriter:
|
|
def __init__(self, db_path: Path = ZOTERO_DB):
|
|
self.db_path = db_path
|
|
|
|
def _backup(self):
|
|
ts = time.strftime("%Y%m%d_%H%M%S")
|
|
backup = self.db_path.with_suffix(f".sqlite.backup.{ts}")
|
|
shutil.copy2(str(self.db_path), str(backup))
|
|
return backup
|
|
|
|
def _get_next_item_id(self, conn: sqlite3.Connection) -> int:
|
|
cur = conn.execute("SELECT MAX(itemID) FROM items")
|
|
return (cur.fetchone()[0] or 0) + 1
|
|
|
|
def _get_field_id(self, conn: sqlite3.Connection, field_name: str) -> int:
|
|
cur = conn.execute("SELECT fieldID FROM fields WHERE fieldName = ?", (field_name,))
|
|
row = cur.fetchone()
|
|
if row:
|
|
return row[0]
|
|
cur = conn.execute("INSERT INTO fields (fieldName) VALUES (?) RETURNING fieldID", (field_name,))
|
|
return cur.fetchone()[0]
|
|
|
|
def _get_or_create_value(self, conn: sqlite3.Connection, value: str) -> int:
|
|
cur = conn.execute("SELECT valueID FROM itemDataValues WHERE value = ?", (value,))
|
|
row = cur.fetchone()
|
|
if row:
|
|
return row[0]
|
|
cur = conn.execute("INSERT INTO itemDataValues (value) VALUES (?) RETURNING valueID", (value,))
|
|
return cur.fetchone()[0]
|
|
|
|
def add_preprint(self, title: str, authors: List[str], url: str,
|
|
arxiv_id: Optional[str] = None, year: Optional[str] = None,
|
|
collection_name: str = "Research Stack") -> str:
|
|
import random, string
|
|
key = "".join(random.choices(string.ascii_uppercase + string.digits, k=8))
|
|
backup_path = self._backup()
|
|
try:
|
|
with sqlite3.connect(str(self.db_path)) as conn:
|
|
item_id = self._get_next_item_id(conn)
|
|
cur = conn.execute("SELECT itemTypeID FROM itemTypes WHERE typeName = 'preprint'")
|
|
row = cur.fetchone()
|
|
preprint_type_id = row[0] if row else 22
|
|
conn.execute(
|
|
"""INSERT INTO items (itemID, itemTypeID, dateAdded, dateModified, clientDateModified, libraryID, key, version, synced)
|
|
VALUES (?, ?, datetime('now'), datetime('now'), datetime('now'), 1, ?, 0, 0)""",
|
|
(item_id, preprint_type_id, key),
|
|
)
|
|
meta = {"title": title, "url": url, "extra": arxiv_id or ""}
|
|
if year:
|
|
meta["date"] = year
|
|
for field_name, val in meta.items():
|
|
if not val:
|
|
continue
|
|
fid = self._get_field_id(conn, field_name)
|
|
vid = self._get_or_create_value(conn, val)
|
|
conn.execute("INSERT INTO itemData (itemID, fieldID, valueID) VALUES (?,?,?)",
|
|
(item_id, fid, vid))
|
|
for i, author in enumerate(authors[:20]):
|
|
author = author.strip()
|
|
if not author:
|
|
continue
|
|
if "," in author:
|
|
parts = [p.strip() for p in author.split(",", 1)]
|
|
last, first = parts[0], parts[1] if len(parts) > 1 else ""
|
|
else:
|
|
parts = author.split()
|
|
last = parts[-1] if parts else ""
|
|
first = " ".join(parts[:-1]) if len(parts) > 1 else ""
|
|
cur = conn.execute(
|
|
"SELECT creatorID FROM creators WHERE firstName = ? AND lastName = ? AND fieldMode = ?",
|
|
(first, last, 0),
|
|
)
|
|
row = cur.fetchone()
|
|
if row:
|
|
creator_id = row[0]
|
|
else:
|
|
cur = conn.execute(
|
|
"INSERT INTO creators (firstName, lastName, fieldMode) VALUES (?,?,?) RETURNING creatorID",
|
|
(first, last, 0),
|
|
)
|
|
creator_id = cur.fetchone()[0]
|
|
conn.execute(
|
|
"INSERT INTO itemCreators (itemID, creatorID, creatorTypeID, orderIndex) VALUES (?,?,1,?)",
|
|
(item_id, creator_id, i),
|
|
)
|
|
cur = conn.execute("SELECT collectionID FROM collections WHERE collectionName = ?", (collection_name,))
|
|
row = cur.fetchone()
|
|
if row:
|
|
col_id = row[0]
|
|
else:
|
|
cur = conn.execute(
|
|
"""INSERT INTO collections (collectionName, parentCollectionID, libraryID, key, version, synced, clientDateModified)
|
|
VALUES (?, NULL, 1, ?, 0, 0, datetime('now')) RETURNING collectionID""",
|
|
(collection_name, key),
|
|
)
|
|
col_id = cur.fetchone()[0]
|
|
cur = conn.execute(
|
|
"SELECT COALESCE(MAX(orderIndex), -1) + 1 FROM collectionItems WHERE collectionID = ?",
|
|
(col_id,),
|
|
)
|
|
order_idx = cur.fetchone()[0]
|
|
conn.execute(
|
|
"INSERT INTO collectionItems (collectionID, itemID, orderIndex) VALUES (?,?,?)",
|
|
(col_id, item_id, order_idx),
|
|
)
|
|
conn.commit()
|
|
return key
|
|
except Exception:
|
|
shutil.copy2(str(backup_path), str(self.db_path))
|
|
raise
|
|
|
|
# ── Watcher State ─────────────────────────────────────────────────────────────
|
|
class WatcherState:
|
|
def __init__(self, path: Path = STATE_FILE):
|
|
self.path = path
|
|
self.data: Dict = self._load()
|
|
|
|
def _load(self) -> Dict:
|
|
if self.path.exists():
|
|
with open(self.path) as f:
|
|
return json.load(f)
|
|
return {"processed_hashes": [], "last_run": None}
|
|
|
|
def save(self):
|
|
with open(self.path, "w") as f:
|
|
json.dump(self.data, f, indent=2)
|
|
|
|
def is_new(self, file_hash: str) -> bool:
|
|
return file_hash not in self.data.get("processed_hashes", [])
|
|
|
|
def mark_processed(self, file_hash: str):
|
|
self.data.setdefault("processed_hashes", []).append(file_hash)
|
|
self.data["last_run"] = time.strftime("%Y-%m-%d %H:%M:%S")
|
|
self.save()
|
|
|
|
# ── PDF Analyzer ────────────────────────────────────────────────────────────
|
|
class PDFAnalyzer:
|
|
ARXIV_RE = re.compile(r"(\d{4}\.\d{4,5}(?:v\d+)?)")
|
|
DOI_RE = re.compile(r"10\.\d{4,9}/[-._;()/:A-Z0-9a-z]+")
|
|
|
|
def analyze(self, path: Path) -> DiscoveredPDF:
|
|
name = path.stem
|
|
arxiv_match = self.ARXIV_RE.search(name)
|
|
arxiv_id = arxiv_match.group(1) if arxiv_match else None
|
|
doi_guess = None
|
|
if not arxiv_id:
|
|
try:
|
|
with open(path, "rb") as f:
|
|
header = f.read(8192).decode("utf-8", errors="ignore")
|
|
dm = self.DOI_RE.search(header)
|
|
if dm:
|
|
doi_guess = dm.group(0)
|
|
am = self.ARXIV_RE.search(header)
|
|
if am:
|
|
arxiv_id = am.group(1)
|
|
except Exception:
|
|
pass
|
|
title_guess = self._clean_title(name)
|
|
h = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
return DiscoveredPDF(path=path, arxiv_id=arxiv_id, doi=doi_guess, title_guess=title_guess, hash=h)
|
|
|
|
@staticmethod
|
|
def _clean_title(filename: str) -> str:
|
|
t = filename.replace("_", " ").replace("-", " ")
|
|
t = re.sub(r"\b\d{4}\.\d{4,5}v?\d*\b", "", t)
|
|
t = re.sub(r"\s+", " ", t).strip()
|
|
return t
|
|
|
|
# ── Ingest Pipeline ─────────────────────────────────────────────────────────
|
|
class IngestPipeline:
|
|
def __init__(self):
|
|
self.analyzer = PDFAnalyzer()
|
|
self.arxiv = ArxivClient()
|
|
self.zotero = ZoteroWriter()
|
|
self.state = WatcherState()
|
|
|
|
def run_once(self, watch_dirs: List[str], dry_run: bool = False) -> List[str]:
|
|
discovered: List[DiscoveredPDF] = []
|
|
for d in watch_dirs:
|
|
root = Path(d)
|
|
if not root.exists():
|
|
continue
|
|
for pdf in root.rglob("*.pdf"):
|
|
# Skip already-archived files
|
|
if CANONICAL_DIR in pdf.parents:
|
|
continue
|
|
info = self.analyzer.analyze(pdf)
|
|
if self.state.is_new(info.hash):
|
|
discovered.append(info)
|
|
|
|
reports: List[str] = []
|
|
for pdf in discovered:
|
|
report = self._process(pdf, dry_run=dry_run)
|
|
reports.append(report)
|
|
if not dry_run:
|
|
self.state.mark_processed(pdf.hash)
|
|
return reports
|
|
|
|
def _process(self, pdf: DiscoveredPDF, dry_run: bool = False) -> str:
|
|
lines = [f"📄 {pdf.path.name}"]
|
|
if pdf.arxiv_id:
|
|
lines.append(f" arXiv ID: {pdf.arxiv_id}")
|
|
if pdf.doi:
|
|
lines.append(f" DOI guess: {pdf.doi}")
|
|
|
|
title = pdf.title_guess or "Unknown"
|
|
authors: List[str] = []
|
|
year: Optional[str] = None
|
|
url = ""
|
|
|
|
if pdf.arxiv_id:
|
|
meta = self.arxiv.fetch_meta(pdf.arxiv_id)
|
|
if meta:
|
|
title = meta["title"]
|
|
authors = meta["authors"]
|
|
year = meta["published"]
|
|
url = meta["pdf_url"]
|
|
lines.append(f" Fetched title: {title}")
|
|
lines.append(f" Authors: {', '.join(authors[:3])}")
|
|
else:
|
|
lines.append(" ⚠️ Could not fetch arXiv metadata")
|
|
|
|
if dry_run:
|
|
lines.append(" ⏸️ DRY RUN — no changes made")
|
|
return "\n".join(lines)
|
|
|
|
# Ingest to Zotero
|
|
try:
|
|
zkey = self.zotero.add_preprint(title, authors, url, arxiv_id=pdf.arxiv_id, year=year)
|
|
lines.append(f" 📥 Zotero key: {zkey}")
|
|
except Exception as e:
|
|
lines.append(f" ⚠️ Zotero ingest failed: {e}")
|
|
|
|
# Move to canonical dir with safe name
|
|
try:
|
|
safe_title = self._safe_slug(title)
|
|
dest_name = f"{pdf.arxiv_id or 'unknown'}_{safe_title}.pdf"
|
|
dest = CANONICAL_DIR / dest_name
|
|
shutil.move(str(pdf.path), str(dest))
|
|
lines.append(f" 💾 Archived to: {dest}")
|
|
except Exception as e:
|
|
lines.append(f" ⚠️ Move failed: {e}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
@staticmethod
|
|
def _safe_slug(title: str) -> str:
|
|
replacements = {
|
|
"\\": " ", "$": "", "^": "", "_": " ", "{": "", "}": "",
|
|
"~": " ", "%": "pct", "&": "and", "#": "", "@": "at",
|
|
}
|
|
for old, new in replacements.items():
|
|
title = title.replace(old, new)
|
|
title = re.sub(r"[^\w\s-]", "", title)
|
|
title = re.sub(r"[-\s]+", "_", title).strip("_")
|
|
return title[:80] if title else "paper"
|
|
|
|
def run_daemon(self, watch_dirs: List[str], interval: int = CHECK_INTERVAL):
|
|
self._log(f"👁️ Ingest Watcher daemon started")
|
|
self._log(f" Watching: {', '.join(watch_dirs)}")
|
|
self._log(f" Archive: {CANONICAL_DIR}")
|
|
self._log(f" Interval: {interval}s")
|
|
while True:
|
|
reports = self.run_once(watch_dirs)
|
|
if reports:
|
|
for r in reports:
|
|
self._log(r)
|
|
time.sleep(interval)
|
|
|
|
@staticmethod
|
|
def _log(msg: str):
|
|
ts = time.strftime("%Y-%m-%d %H:%M:%S")
|
|
line = f"[{ts}] {msg}"
|
|
print(line)
|
|
with open(LOG_FILE, "a") as f:
|
|
f.write(line + "\n")
|
|
|
|
# ── Main ─────────────────────────────────────────────────────────────────────
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Smart Ingest Watcher")
|
|
parser.add_argument("--watch", nargs="*", default=DEFAULT_WATCH_DIRS, help="Directories to watch")
|
|
parser.add_argument("--once", action="store_true", help="Run one scan and exit")
|
|
parser.add_argument("--dry-run", action="store_true", help="Don't modify anything")
|
|
parser.add_argument("--daemon", action="store_true", help="Run forever")
|
|
parser.add_argument("--interval", type=int, default=CHECK_INTERVAL, help="Seconds between scans")
|
|
args = parser.parse_args()
|
|
|
|
pipeline = IngestPipeline()
|
|
if args.once:
|
|
reports = pipeline.run_once(args.watch, dry_run=args.dry_run)
|
|
for r in reports:
|
|
print(r)
|
|
print(f"\nProcessed {len(reports)} new PDFs.")
|
|
elif args.daemon:
|
|
pipeline.run_daemon(args.watch, interval=args.interval)
|
|
else:
|
|
parser.print_help()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|