72 lines
2.0 KiB
Python
72 lines
2.0 KiB
Python
import argparse
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
from backend import config
|
|
from backend.database import init_db
|
|
from backend.news_service import process_and_store_news
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(prog="clawfort", description="ClawFort operations CLI")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
force_fetch_parser = subparsers.add_parser(
|
|
"force-fetch",
|
|
help="Run one immediate news fetch cycle",
|
|
description="Trigger one immediate news fetch run outside scheduler cadence.",
|
|
)
|
|
force_fetch_parser.set_defaults(handler=handle_force_fetch)
|
|
|
|
return parser
|
|
|
|
|
|
def validate_runtime() -> None:
|
|
if not config.PERPLEXITY_API_KEY and not config.OPENROUTER_API_KEY:
|
|
raise RuntimeError(
|
|
"No provider API key configured. Set PERPLEXITY_API_KEY or OPENROUTER_API_KEY in the environment."
|
|
)
|
|
|
|
|
|
def handle_force_fetch(_: argparse.Namespace) -> int:
|
|
start = time.monotonic()
|
|
|
|
try:
|
|
validate_runtime()
|
|
os.makedirs("data", exist_ok=True)
|
|
init_db()
|
|
|
|
stored_count = asyncio.run(process_and_store_news())
|
|
elapsed = time.monotonic() - start
|
|
|
|
print(f"force-fetch succeeded: stored={stored_count} elapsed={elapsed:.1f}s")
|
|
return 0
|
|
except Exception as exc:
|
|
logger.exception("force-fetch failed")
|
|
print(f"force-fetch failed: {exc}", file=sys.stderr)
|
|
print(
|
|
"Check API keys, network connectivity, and provider status, then retry the command.",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
handler = args.handler
|
|
return handler(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|