upload_logs_to_oss.py 1.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940
  1. #!/usr/bin/env python3
  2. """Upload a log file from logs/ to OSS (manual folder, no MySQL).
  3. Usage:
  4. python scripts/upload_logs_to_oss.py run_xxx.log
  5. # 从 Python 调用
  6. from scripts.upload_logs_to_oss import main
  7. oss_path = main("run_xxx.log")
  8. """
  9. from __future__ import annotations
  10. import sys
  11. from pathlib import Path
  12. _ROOT = Path(__file__).resolve().parents[1]
  13. _LOGS_DIR = _ROOT / "logs"
  14. if str(_ROOT) not in sys.path:
  15. sys.path.insert(0, str(_ROOT))
  16. from supply_infra.config import get_infra_settings
  17. from supply_infra.oss.client import OssClient
  18. def main(filename: str) -> str:
  19. """Read ``logs/{filename}``, upload to OSS, return public OSS path."""
  20. local_path = _LOGS_DIR / Path(filename).name
  21. if not local_path.is_file():
  22. raise FileNotFoundError(f"Log file not found: {local_path}")
  23. settings = get_infra_settings()
  24. client = OssClient(settings, root_prefix=settings.aliyun_oss_manual_log_prefix)
  25. object_key = client.object_key(local_path.name)
  26. return client.upload_file(local_path, object_key)
  27. if __name__ == "__main__":
  28. oss_path = main("run_generate_demand_agent_20260716_160355_8f751d0d.html")
  29. print(oss_path)