自動化：スケジューリングと業務改善の型｜「再実行安全×観測可能×静かに動く」を仕組みにする

auto_job/
  app/
    __init__.py
    main.py          # 入口（引数パース/ワイヤリング）
    job.py           # 業務ロジック本体（純粋関数を中心に）
    io_utils.py      # 原子置換/パス/ロック
    logconf.py       # JSONロガー＋相関ID（任意）
  config/
    config.yml       # 入出力パス/実行時設定
  data/
    in/              # 入力（読み取り専用）
    work/            # 中間（Parquet/キャッシュ）
    out/             # 出力（原子置換で置き換え）
  tests/
    test_job.py

from pathlib import Path
import tempfile, os, time

class FileLock:
def **init**(self, path: Path, timeout: float = 0):
self.path, self.timeout = path, timeout
self.\_fd = None
def **enter**(self):
start = time.time()
while True:
try:
self.\_fd = os.open(self.path, os.O\_CREAT | os.O\_EXCL | os.O\_RDWR)
os.write(self.\_fd, str(os.getpid()).encode())
return self
except FileExistsError:
if time.time() - start > self.timeout:
raise TimeoutError(f"lock busy: {self.path}")
time.sleep(0.5)
def **exit**(self, \*exc):
try:
os.close(self.\_fd)
os.remove(self.path)
except Exception:
pass

def atomic\_write\_text(path: Path, text: str, encoding: str = "utf-8") -> None:
path.parent.mkdir(parents=True, exist\_ok=True)
with tempfile.NamedTemporaryFile("w", delete=False, dir=path.parent, encoding=encoding, newline="") as tmp:
tmp.write(text)
tmp\_path = Path(tmp.name)
os.replace(tmp\_path, path)

from pathlib import Path
import pandas as pd

def make\_report(src: Path, out\_xlsx: Path) -> dict:
df = pd.read\_csv(src, parse\_dates=\["date"], dtype={"store":"string"})
g = df.groupby(\[df\["date"].dt.date, "store"]).agg(sales=("sales","sum"))
g = g.reset\_index().rename(columns={"date":"day"})
with pd.ExcelWriter(out\_xlsx, engine="xlsxwriter") as w:
g.to\_excel(w, sheet\_name="日次売上", index=False)
return {"rows": int(len(df)), "groups": int(len(g))}

import argparse, json
from pathlib import Path
from datetime import datetime
from app.io_utils import FileLock
from app.job import make_report

DEF\_IN = Path("data/in/sales.csv")
DEF\_OUT = Path("data/out/report.xlsx")

parser = argparse.ArgumentParser()
parser.add\_argument("--src", type=Path, default=DEF\_IN)
parser.add\_argument("--out", type=Path, default=DEF\_OUT)
parser.add\_argument("--dry-run", action="store\_true")

if **name** == "**main**":
args = parser.parse\_args()
with FileLock(Path("data/.report.lock"), timeout=3):
stats = make\_report(args.src, args.out)
print(json.dumps({
"ts": datetime.now().isoformat(timespec="seconds"),
"rows": stats\["rows"], "groups": stats\["groups"],
"out": str(args.out)
}, ensure\_ascii=False))

# 毎朝 6:10 に実行、ログは日付付きで保存
10 6 * * * cd /path/to/auto_job && /usr/bin/python -m app.main \
  >> logs/$(date +\%Y\%m\%d).log 2>&1

schtasks /Create /SC DAILY /ST 06:10 /TN AutoReport ^
 /TR "cmd /c cd C:\path\auto_job && py -3 -m app.main >> logs\%date:~0,10%.log 2>&1"

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime
from pathlib import Path
from app.job import make_report

sched = BlockingScheduler(timezone="Asia/Tokyo")

@sched.scheduled\_job(CronTrigger(hour=6, minute=10))
def daily():
src = Path("data/in/sales.csv")
out = Path("data/out/report.xlsx")
info = make\_report(src, out)
print({"ts": datetime.now().isoformat(), \*\*info})

if **name** == "**main**":
sched.start()

name: nightly
on:
  schedule:
    - cron: "10 21 * * *"  # JST 6:10（UTC-9時間）
jobs:
  run:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.12' }
      - run: pip install -r requirements.txt
      - run: python -m app.main --src data/in/sales.csv --out data/out/report.xlsx
      - uses: actions/upload-artifact@v4
        with: { name: report, path: data/out/report.xlsx }

import smtplib, os
from email.message import EmailMessage
import json, urllib.request

def send\_mail(subject: str, body: str) -> None:
host = os.environ.get("SMTP\_HOST")
to = os.environ.get("MAIL\_TO")
if not host or not to:
return
msg = EmailMessage()
msg\["Subject"], msg\["From"], msg\["To"] = subject, to, to
msg.set\_content(body)
with smtplib.SMTP(host) as s:
s.send\_message(msg)

def send\_slack(text: str) -> None:
url = os.environ.get("SLACK\_WEBHOOK")
if not url:
return
data = json.dumps({"text": text}).encode()
req = urllib.request.Request(url, data=data, headers={"Content-Type":"application/json"})
urllib.request.urlopen(req, timeout=5)