#!/usr/bin/env python3
import argparse
import logging
import os
import sys
from pathlib import Path
import maintain
def configure_logging(log_directory: Path) -> None:
log_directory.mkdir(parents=True, exist_ok=True)
from logging.handlers import RotatingFileHandler
file_handler = RotatingFileHandler(
log_directory / "automation.log",
maxBytes=5_000_000,
backupCount=5,
encoding="utf-8",
)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s UTC %(levelname)s %(message)s",
handlers=[
file_handler,
logging.StreamHandler(sys.stdout),
],
)
# Ensure the timestamps labeled UTC actually use UTC.
import time
logging.Formatter.converter = time.gmtime
def run_maintenance(
app_directory: Path,
public_directory: Path,
) -> int:
config_path = app_directory / "sources.json"
database_path = app_directory / "hustlefinder.sqlite3"
if not config_path.is_file():
logging.error("Missing configuration: %s", config_path)
return 1
public_directory.mkdir(parents=True, exist_ok=True)
# The existing module reads these globals during its operations.
maintain.CONFIG = str(config_path)
maintain.DB = str(database_path)
logging.info("Maintenance started")
try:
configuration = maintain.config()
if not isinstance(configuration.get("sources"), list):
raise ValueError("'sources' must be an array")
if not isinstance(
configuration.get("allowed_link_hosts", []), list
):
raise ValueError("'allowed_link_hosts' must be an array")
with maintain.database():
pass
except Exception:
logging.exception("Configuration or database initialization failed")
return 1
failed = False
# Continue other maintenance even when a feed fails.
try:
failures = maintain.refresh()
if failures:
failed = True
logging.error(
"%d feed refresh(es) failed; inspect the runs table",
failures,
)
else:
logging.info("Feed refresh completed")
except Exception:
failed = True
logging.exception("Feed refresh failed")
try:
maintain.check()
logging.info("Application-link checks completed")
except Exception:
failed = True
logging.exception("Application-link checks failed")
try:
output_path = public_directory / "jobs.json"
# Existing export uses a temporary file followed by replacement.
# Only approved/published records enter the public export.
maintain.export(str(output_path), public=True)
logging.info("Published job data exported to %s", output_path)
except Exception:
failed = True
logging.exception("Public export failed")
logging.info(
"Maintenance finished: %s",
"completed with errors" if failed else "successful",
)
return 1 if failed else 0
def main() -> int:
parser = argparse.ArgumentParser(
description="Run HustleFinder listing maintenance"
)
parser.add_argument(
"--app-dir",
type=Path,
default=Path(__file__).resolve().parent,
)
parser.add_argument(
"--public-dir",
type=Path,
required=True,
)
args = parser.parse_args()
app_directory = args.app_dir.resolve()
public_directory = args.public_dir.resolve()
# Private files should not become publicly served files.
if (
public_directory == app_directory
or public_directory in app_directory.parents
):
parser.error(
"--public-dir must not be the app directory "
"or a parent of the app directory"
)
os.umask(0o027)
configure_logging(app_directory / "logs")
try:
return run_maintenance(app_directory, public_directory)
except Exception:
logging.exception("Unexpected maintenance failure")
return 1
if __name__ == "__main__":
raise SystemExit(main())