“落ちない”社内自動化3選：再実行安全・ロック・JSONログで回す設計とテンプレ

from pathlib import Path
import tempfile, os, time, json, logging
from logging.handlers import TimedRotatingFileHandler

class FileLock:
def **init**(self, path: Path, timeout: float = 3):
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="utf-8"):
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)

# JSON構造化ロガー（回転）

log = logging.getLogger("job"); log.setLevel(logging.INFO)
handler = TimedRotatingFileHandler("logs/job.log", when="midnight", backupCount=7, encoding="utf-8")
class JsonFmt(logging.Formatter):
def format(self, r: logging.LogRecord):
payload = {"ts": self.formatTime(r, "%Y-%m-%dT%H:%M:%S%z"), "lvl": r.levelname, "msg": r.getMessage(), "module": r.module, "line": r.lineno}
if r.exc\_info:
payload\["exc"] = self.formatException(r.exc\_info)
return json.dumps(payload, ensure\_ascii=False)
handler.setFormatter(JsonFmt()); log.addHandler(handler)

import argparse
from pathlib import Path
import pandas as pd
from utils import FileLock, log

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

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

if **name** == "**main**":
ap = argparse.ArgumentParser()
ap.add\_argument("--src", type=Path, default=DEF\_IN)
ap.add\_argument("--out", type=Path, default=DEF\_OUT)
a = ap.parse\_args()
with FileLock(Path(".report.lock"), timeout=3):
info = build\_report(a.src, a.out)
log.info(f"report done rows={info\['rows']} groups={info\['groups']} out={a.out}")
print(info)

from __future__ import annotations
from pathlib import Path
import json, os, time
import requests
from utils import FileLock, log

BASE = "[https://api.example.com](https://api.example.com)"  # 実運用は環境変数で
API\_KEY = os.getenv("API\_KEY", "")
HEADERS = {"User-Agent":"pythonbunseki-monitor/1.0","Accept":"application/json","X-API-Key"\:API\_KEY}

HIST = Path("data/history.ndjson")
SEEN = set()  # 簡易（本番は過去ファイルを読む）

def fetch\_items() -> list\[dict]:
r = requests.get(f"{BASE}/items", headers=HEADERS, timeout=10)
if r.status\_code == 429:
raise RuntimeError("rate limited")
r.raise\_for\_status()
return r.json().get("items", \[])

def diff\_and\_notify(items: list\[dict]):
new = \[]
for it in items:
k = (it.get("id"), it.get("price"), it.get("stock"))
if k not in SEEN:
SEEN.add(k); new\.append(it)
if not new:
log.info("no changes"); return
\# 保存（NDJSON）
HIST.parent.mkdir(parents=True, exist\_ok=True)
with HIST.open("a", encoding="utf-8") as f:
for it in new:
f.write(json.dumps(it, ensure\_ascii=False) + "\n")
\# 通知（ダミー：標準出力）
for it in new\[:10]:
log.info(f"change id={it\['id']} price={it.get('price')} stock={it.get('stock')}")

if **name** == "**main**":
with FileLock(Path(".monitor.lock"), timeout=3):
tries = 0
while True:
try:
items = fetch\_items(); diff\_and\_notify(items); break
except (requests.Timeout, requests.ConnectionError) as e:
tries += 1
if tries >= 3: raise
wait = min(8, 0.5 \* 2 \*\* (tries-1))
log.warning(f"retry {tries} in {wait}s: {e}"); time.sleep(wait)

from pathlib import Path
import pandas as pd
from datetime import datetime
from utils import FileLock, log

SRC\_DIR = Path("data/source")
DST\_DIR = Path("backup")
KEEP = 30

def backup\_csvs():
today = datetime.now().strftime("%Y%m%d")
out = DST\_DIR/today; out.mkdir(parents=True, exist\_ok=True)
for p in sorted(SRC\_DIR.glob("*.csv")):
df = pd.read\_csv(p)
df.to\_parquet(out/(p.stem+".parquet"), compression="zstd")
\# 古い世代を削除
gens = sorted(\[d for d in DST\_DIR.iterdir() if d.is\_dir()])
for d in gens\[:-KEEP]:
for f in d.glob("\*\*/*"): f.unlink()
d.rmdir()
log.info(f"backup done dir={out} keep={KEEP}")

if **name** == "**main**":
with FileLock(Path(".backup.lock"), timeout=3):
backup\_csvs()