97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
import fcntl
|
|
import json
|
|
import logging
|
|
import os
|
|
import random
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ArticleRegistry:
|
|
def __init__(self, path: Path):
|
|
self.path = path
|
|
|
|
def _lock(self) -> tuple:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
fd = os.open(str(self.path), os.O_RDWR | os.O_CREAT)
|
|
fcntl.flock(fd, fcntl.LOCK_EX)
|
|
data = {}
|
|
if os.path.getsize(str(self.path)) > 0:
|
|
chunks = []
|
|
while True:
|
|
chunk = os.read(fd, 65536)
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
raw = b"".join(chunks).decode(errors="replace")
|
|
if raw.strip():
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
data = {}
|
|
if not isinstance(data, dict):
|
|
data = {}
|
|
now = datetime.now()
|
|
kept: dict = {}
|
|
for title, meta in data.items():
|
|
if not isinstance(meta, dict):
|
|
continue
|
|
ts = meta.get("time", "")
|
|
try:
|
|
age_days = (
|
|
now - datetime.fromisoformat(ts)
|
|
).total_seconds() / 86400
|
|
except (ValueError, TypeError):
|
|
age_days = 0
|
|
if age_days < 7:
|
|
kept[title] = meta
|
|
return fd, kept
|
|
except Exception as e:
|
|
logger.warning("Failed to lock article registry: %s", e)
|
|
return None, {}
|
|
|
|
@staticmethod
|
|
def _unlock(fd) -> None:
|
|
if fd is not None:
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_UN)
|
|
os.close(fd)
|
|
except Exception as e:
|
|
logger.debug("Failed to unlock article registry: %s", e)
|
|
|
|
def reserve(self, title: str, owner: str) -> bool:
|
|
fd, registry = self._lock()
|
|
if fd is None:
|
|
return True
|
|
try:
|
|
if title in registry:
|
|
return False
|
|
registry[title] = {"bot": owner, "time": datetime.now().isoformat()}
|
|
self.path.write_text(json.dumps(registry, indent=2, default=str))
|
|
return True
|
|
finally:
|
|
self._unlock(fd)
|
|
|
|
def reserve_unused(self, articles: list[dict], owner: str, clean) -> Optional[dict]:
|
|
if not articles:
|
|
return None
|
|
fd, registry = self._lock()
|
|
if fd is None:
|
|
return random.choice(articles) if articles else None
|
|
try:
|
|
used = set(registry.keys())
|
|
random.shuffle(articles)
|
|
for a in articles:
|
|
title = clean(a.get("title", "")[:200])
|
|
if title and title not in used:
|
|
registry[title] = {"bot": owner, "time": datetime.now().isoformat()}
|
|
self.path.write_text(json.dumps(registry, indent=2, default=str))
|
|
return a
|
|
return None
|
|
finally:
|
|
self._unlock(fd)
|