請求書PDF SharePoint自動配置

Issue #123 | 2026-03-17 | architect

1. やりたいこと

MFクラウドから取得した請求書PDFを、SharePointの所定フォルダに正しいファイル名で自動配置する。現在は手動でリネーム＆アップロードしているが、これをパイプラインの1ステップとして自動化する。

2. 設計方針

2.1 請求書番号 → 担当者名のマッピング

請求書番号（billing_number）と担当者名の対応は、JSONの設定ファイルで管理する。

設定ファイルの場所: ops/invoice/data/billing_map.json

{
  "partner_display": "TFHD",
  "company_name": "合同会社MNML",
  "sp_base_path": "10_Corporate/accounting/30_売上請求/TFHD",
  "billings": {
    "11": "(内山・根崎)",
    "12": "(小川)"
  }
}

2.2 アップロード方式: Graph API 直接アップロード

経費精算パイプライン（accounting/export_excel.py）と同じ方式を採用する。

2.3 冪等性

Graph API の PUT は上書き動作（同名ファイルが存在すれば更新される）。これは冪等。

• 同じ月のパイプラインを何度実行しても安全
• SharePoint側にバージョン履歴が自動で残るため、誤って上書きしても復元可能
• ローカルのダウンロード済みPDFは既存の download ステップが冪等にスキップする

2.4 BKディレクトリへのバックアップ

自動化しない。理由:

• SharePoint自体がバージョン管理機能を持っている（ファイル履歴から復元可能）
• BKディレクトリは過去の手動運用の名残であり、二重管理になる
• 必要であれば将来追加は容易（PUTをもう1回呼ぶだけ）

3. ファイル構成

4. 処理フロー

4.1 全体パイプライン（変更後）

4.2 upload ステップの内部処理

5. インターフェース定義

5.1 データモデル: BillingMap

# ops/invoice/pipeline/steps/upload.py

from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path

_DATA_DIR = Path(__file__).resolve().parent.parent.parent / "data"
_BILLING_MAP_PATH = _DATA_DIR / "billing_map.json"

# SharePoint Drive ID（経費精算と同じ Communication site のドキュメント）
_SP_DRIVE_ID = "b!tNcjqNNMOkCi_YGmMqT7kS1WUEYULklGuffT9GtBXJBc4QvplsmnSJQywtJi22rt"


@dataclass
class BillingMap:
    """請求書番号と担当者名の対応表。"""
    partner_display: str          # "TFHD"
    company_name: str             # "合同会社MNML"
    sp_base_path: str             # "10_Corporate/accounting/30_売上請求/TFHD"
    billings: dict[str, str]      # {"11": "(内山・根崎)", "12": "(小川)"}

    @classmethod
    def load(cls) -> BillingMap:
        """billing_map.json を読み込む。"""
        raw = json.loads(_BILLING_MAP_PATH.read_text(encoding="utf-8"))
        return cls(
            partner_display=raw["partner_display"],
            company_name=raw["company_name"],
            sp_base_path=raw["sp_base_path"],
            billings=raw["billings"],
        )

    def resolve_filename(self, billing_number: str, yyyymm: str) -> str | None:
        """請求書番号からSharePoint用のファイル名を生成する。

        対応するエントリがなければ None を返す。
        """
        suffix = self.billings.get(billing_number)
        if suffix is None:
            return None
        return f"請求書{yyyymm}_{self.partner_display}_{self.company_name}{suffix}.pdf"

    def resolve_remote_path(self, billing_number: str, yyyymm: str, year: int) -> str | None:
        """SharePoint上のフルパス（Drive root からの相対パス）を返す。"""
        filename = self.resolve_filename(billing_number, yyyymm)
        if filename is None:
            return None
        return f"{self.sp_base_path}/{year}/{filename}"

5.2 メイン関数: run()

async def run(cfg: PipelineConfig) -> None:
    """ローカルの請求書PDFをリネームしてSharePointにアップロードする。"""
    # 1. billing_map.json 読み込み
    bmap = BillingMap.load()

    # 2. ダウンロード済みPDF一覧
    pdf_files = sorted(cfg.invoice_dir.glob("*.pdf"))
    if not pdf_files:
        console.print(f"[yellow]アップロード対象なし: {cfg.invoice_dir}[/]")
        return

    # 3. 認証準備
    from onedrive.app.auth import OneDriveAuth
    auth = OneDriveAuth()

    uploaded = 0
    skipped = 0
    for pdf_path in pdf_files:
        # 4. ファイル名から billing_number を抽出（先頭の"_"区切り）
        billing_number = pdf_path.stem.split("_")[0]

        # 5. SharePoint用パスを解決
        remote_path = bmap.resolve_remote_path(billing_number, cfg.yyyymm, cfg.year)
        if remote_path is None:
            console.print(f"[yellow]スキップ（マッピングなし）: {pdf_path.name} "
                          f"(billing_number={billing_number})[/]")
            skipped += 1
            continue

        sp_filename = remote_path.rsplit("/", 1)[-1]

        if cfg.dry_run:
            console.print(f"[dim][DRY-RUN] {pdf_path.name} → {sp_filename}[/]")
            continue

        # 6. Graph API PUT でアップロード
        try:
            link = await _upload_pdf(auth, pdf_path, remote_path)
            console.print(f"[green]アップロード:[/] {pdf_path.name} → {sp_filename}")
            console.print(f"  [dim]{link}[/]")
            uploaded += 1
        except Exception as e:
            console.print(f"[red]失敗:[/] {pdf_path.name} — {e}")

    console.print(f"\nアップロード完了: {uploaded} 件, スキップ: {skipped} 件")
    if cfg.dry_run:
        console.print("[yellow]--dry-run のため実際のアップロードは行われていません。[/]")

5.3 アップロード関数: _upload_pdf()

async def _upload_pdf(auth, pdf_path: Path, remote_path: str) -> str:
    """PDFをSharePointにアップロードし、共有リンクを返す。

    Graph API: PUT /drives/{driveId}/root:/{path}:/content
    冪等: 同名ファイルがあれば上書き（バージョン履歴は自動保存）。
    """
    headers = await auth.get_auth_header()
    headers["Content-Type"] = "application/octet-stream"

    url = (f"https://graph.microsoft.com/v1.0/drives/{_SP_DRIVE_ID}"
           f"/root:/{remote_path}:/content")

    data = pdf_path.read_bytes()

    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.put(url, headers=headers, content=data)
        resp.raise_for_status()
        item_id = resp.json()["id"]

    # 共有リンク作成
    link_headers = await auth.get_auth_header()
    link_headers["Content-Type"] = "application/json"
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.post(
            f"https://graph.microsoft.com/v1.0/drives/{_SP_DRIVE_ID}/items/{item_id}/createLink",
            headers=link_headers,
            json={"type": "view", "scope": "organization"},
        )
        resp.raise_for_status()
        return resp.json().get("link", {}).get("webUrl", "")

6. ファイル名変換ロジック

6.1 変換ルール

6.2 変換の仕組み

1. ローカルファイル名 {billing_number}_{partner_name}.pdf から billing_number を抽出（"_" で分割した先頭要素）
2. billing_map.json の billings から billing_number をキーに担当者サフィックスを取得
3. テンプレート: 請求書{YYYYMM}_{partner_display}_{company_name}{suffix}.pdf

6.3 マッピングに存在しない請求書番号の場合

アップロードをスキップし、警告をコンソールに出力する。無理にアップロードしない（ファイル名が不正になるため）。

7. SharePoint配置先

8. CLI変更

8.1 cli.py に upload サブコマンドを追加

# build_parser() 内に追加
sub.add_parser("upload", help="SharePointにPDFを配置")

# main() 内に追加
elif args.step == "upload":
    from invoice.pipeline.steps import upload
    asyncio.run(upload.run(cfg))

使い方:

# 手動実行
python -m invoice.pipeline upload
python -m invoice.pipeline upload --dry-run
python -m invoice.pipeline --year 2026 --month 2 upload

8.2 scheduler/run.py に upload を追加

async def _run_pipeline(log: logging.Logger) -> None:
    from invoice.pipeline.steps import confirm, download, upload

    cfg = PipelineConfig()

    log.info("confirm 開始")
    await confirm.run(cfg)
    log.info("confirm 完了")

    log.info("download 開始")
    await download.run(cfg)
    log.info("download 完了")

    # 新規追加
    log.info("upload 開始")
    await upload.run(cfg)
    log.info("upload 完了")

    log.info("パイプライン完了")

9. 冪等性

10. エラーハンドリング

11. 認証

既存の OneDriveAuth（ops/onedrive/app/auth.py）をそのまま使う。

• Device Code Flow（Mac miniのようなブラウザなし環境向け）
• スコープ: Files.ReadWrite.All, Sites.Read.All（SharePointファイル操作に必要）
• トークン保存: ops/onedrive/.tokens.json（既存）
• 自動リフレッシュ: BaseOAuth が期限切れ時にリフレッシュトークンで更新

新規のOAuth設定や追加スコープは不要。

12. 前提条件・制約

• ops/onedrive/.tokens.json に有効なトークンが存在すること（初回は setup_oauth.py で取得済み）
• Drive ID は経費精算パイプラインと同一の Communication site ドキュメントライブラリ
• billing_map.json は手動メンテナンス。請求書番号の追加・変更時に更新する
• PDF 1ファイルのサイズは Graph API の直接 PUT 上限（4MB）以内であること（請求書PDFは通常100KB未満のため問題なし）

13. 設計のまとめ