MNML 契約管理パイプライン（ops/contract/）

Contract モデル

ステータス遷移

※ auto_renew: true の契約は check 時に expired にせず、Slack通知で「自動更新されるが確認してください」と表示する。

ディレクトリ構成

ops/contract/
├── __init__.py
├── app/
│   ├── __init__.py
│   ├── config.py              # pydantic-settings: SharePointパス、通知設定
│   ├── models.py              # Contract Pydanticモデル
│   ├── repository.py          # contracts.json の読み書き + 公開クエリ関数
│   └── data/
│       └── contracts.json     # 契約マスタデータ
└── pipeline/
    ├── __init__.py
    ├── __main__.py            # python -m contract.pipeline
    ├── cli.py                 # argparse サブコマンド定義
    └── steps/
        ├── __init__.py
        ├── list_contracts.py  # 一覧表示（richテーブル）
        ├── show_contract.py   # 契約詳細表示
        ├── add_contract.py    # 対話的に契約追加
        └── check_expiry.py    # 期限チェック + ステータス更新 + Slack通知

各モジュールの役割

pyproject.toml への追加

# [tool.setuptools.packages.find] の include に追加
include = [..., "contract*"]

# 依存は基本パッケージのみ（pydantic, rich）で追加不要

models.py

from __future__ import annotations

from datetime import date
from typing import Literal

from pydantic import BaseModel, Field


class Contract(BaseModel):
    """契約データモデル。"""

    id: str
    type: Literal["sales", "purchase"]
    counterparty: str
    counterparty_full_name: str | None = None
    counterparty_email: str | None = None
    vdu: list[str]
    contract_type: str
    amount: int = Field(gt=0, description="月額金額（円・税抜）")
    currency: str = "JPY"
    period_start: date
    period_end: date | None = None
    auto_renew: bool = False
    renewal_notice_days: int = 30
    status: Literal["active", "expired", "terminated"]
    payment_cycle: str | None = None
    report_due_day: int | None = Field(default=None, ge=1, le=31)
    sharepoint_contract_path: str | None = None
    sharepoint_deliverable_path: str | None = None
    invoice_folder: str | None = None
    notes: str | None = None

    @property
    def is_expiring_soon(self) -> bool:
        """更新通知期間内かどうか。"""
        if self.period_end is None or self.status != "active":
            return False
        return (self.period_end - date.today()).days <= self.renewal_notice_days

    @property
    def days_until_expiry(self) -> int | None:
        """期限切れまでの日数。無期限ならNone。"""
        if self.period_end is None:
            return None
        return (self.period_end - date.today()).days

    @property
    def is_expired(self) -> bool:
        """期限を過ぎているか。"""
        if self.period_end is None:
            return False
        return date.today() > self.period_end

config.py

from __future__ import annotations

from pathlib import Path

from pydantic_settings import BaseSettings


class ContractSettings(BaseSettings):
    """契約管理パイプラインの設定。"""

    # SharePoint
    contract_sharepoint_base: str = "/10_Corporate/contracts"

    # Slack通知
    slack_bot_token: str = ""
    contract_notify_channel: str = ""  # 通知先チャンネルID

    # デフォルト値
    default_renewal_notice_days: int = 30

    model_config = {"env_file": str(Path(__file__).resolve().parents[3] / ".env")}


# データファイルパス
DATA_DIR: Path = Path(__file__).resolve().parent / "data"
CONTRACTS_JSON: Path = DATA_DIR / "contracts.json"

settings = ContractSettings()

repository.py — 公開クエリ関数

from __future__ import annotations

import json
from pathlib import Path

from contract.app.config import CONTRACTS_JSON
from contract.app.models import Contract


def _load_contracts(path: Path = CONTRACTS_JSON) -> list[Contract]:
    """contracts.json を読み込み、Contractリストを返す。"""
    if not path.exists():
        return []
    with path.open(encoding="utf-8") as f:
        data = json.load(f)
    return [Contract.model_validate(item) for item in data]


def _save_contracts(contracts: list[Contract], path: Path = CONTRACTS_JSON) -> None:
    """Contractリストを contracts.json に書き込む。"""
    data = [c.model_dump(mode="json") for c in contracts]
    with path.open("w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


# === 公開クエリ関数（他パイプラインから呼ばれる） ===

def get_all_contracts() -> list[Contract]:
    """全契約を返す。"""
    return _load_contracts()


def get_active_contracts(
    *,
    type: str | None = None,
    vdu: str | None = None,
) -> list[Contract]:
    """有効な契約のみ返す。type / vdu でフィルタ可能。"""
    contracts = [c for c in _load_contracts() if c.status == "active"]
    if type is not None:
        contracts = [c for c in contracts if c.type == type]
    if vdu is not None:
        contracts = [c for c in contracts if vdu in c.vdu]
    return contracts


def get_contract_by_id(contract_id: str) -> Contract | None:
    """IDで契約を検索。"""
    for c in _load_contracts():
        if c.id == contract_id:
            return c
    return None


def get_contract_by_counterparty(name: str) -> Contract | None:
    """取引先名で契約を検索（部分一致）。"""
    for c in _load_contracts():
        if name in c.counterparty or (c.counterparty_full_name and name in c.counterparty_full_name):
            return c
    return None


def get_contracts_by_vdu(vdu: str) -> list[Contract]:
    """VDU名で契約を絞り込む。"""
    return [c for c in _load_contracts() if vdu in c.vdu]


def get_monthly_amount(counterparty_name: str) -> int | None:
    """取引先の月額金額を返す。見つからなければNone。"""
    contract = get_contract_by_counterparty(counterparty_name)
    if contract is None or contract.status != "active":
        return None
    return contract.amount


def add_contract(contract: Contract) -> None:
    """契約を追加する。ID重複時はValueError。"""
    contracts = _load_contracts()
    if any(c.id == contract.id for c in contracts):
        msg = f"契約ID {contract.id} は既に存在します"
        raise ValueError(msg)
    contracts.append(contract)
    _save_contracts(contracts)


def update_contract(contract: Contract) -> None:
    """契約を更新する。ID不存在時はValueError。"""
    contracts = _load_contracts()
    for i, c in enumerate(contracts):
        if c.id == contract.id:
            contracts[i] = contract
            _save_contracts(contracts)
            return
    msg = f"契約ID {contract.id} が見つかりません"
    raise ValueError(msg)

コマンド一覧

cli.py シグネチャ

from __future__ import annotations

import argparse
import sys


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="contract.pipeline",
        description="MNML 契約管理パイプライン",
    )
    sub = parser.add_subparsers(dest="command", help="実行するコマンド")

    # list
    p_list = sub.add_parser("list", help="契約一覧を表示")
    p_list.add_argument("--type", choices=["sales", "purchase"], help="契約タイプで絞り込み")
    p_list.add_argument("--vdu", help="VDU名で絞り込み")
    p_list.add_argument("--status", choices=["active", "expired", "terminated"], help="ステータスで絞り込み")

    # show
    p_show = sub.add_parser("show", help="契約詳細を表示")
    p_show.add_argument("contract_id", help="契約ID")

    # add
    sub.add_parser("add", help="対話的に契約を追加")

    # check
    p_check = sub.add_parser("check", help="期限チェック・ステータス更新")
    p_check.add_argument("--notify", action="store_true", help="Slack通知を送信")
    p_check.add_argument("--days", type=int, default=30, help="警告日数（デフォルト: 30）")

    return parser


def main(argv: list[str] | None = None) -> None:
    parser = build_parser()
    args = parser.parse_args(argv)

    if not args.command:
        parser.print_help()
        sys.exit(1)

    if args.command == "list":
        from contract.pipeline.steps.list_contracts import run
        run(type_filter=args.type, vdu_filter=args.vdu, status_filter=args.status)
    elif args.command == "show":
        from contract.pipeline.steps.show_contract import run
        run(contract_id=args.contract_id)
    elif args.command == "add":
        from contract.pipeline.steps.add_contract import run
        run()
    elif args.command == "check":
        from contract.pipeline.steps.check_expiry import run
        run(notify=args.notify, warning_days=args.days)

出力例: list

$ python -m contract.pipeline list
┏━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━┓
┃ ID    ┃ タイプ   ┃ 取引先     ┃ VDU    ┃ 月額(税抜) ┃ 期間                    ┃ ステータス ┃
┡━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━┩
│ C001  │ 売上     │ TFHD       │ TFHD   │ 1,500,000 │ 2026-04-01 ~ 2027-03-31 │ active  │
│ S001  │ 仕入     │ 根崎       │ TFHD   │   500,000 │ 2026-04-01 ~ 2027-03-31 │ active  │
│ S002  │ 仕入     │ 小川哲央   │ TFHD   │   300,000 │ 2026-04-01 ~ 2027-03-31 │ active  │
│ S003  │ 仕入     │ 森屋穣     │ MNML   │   200,000 │ 2026-04-01 ~            │ active  │
│ S004  │ 仕入     │ 小川さとこ │ MNML   │   150,000 │ 2026-04-01 ~ 2026-09-30 │ active  │
│ S005  │ 仕入     │ Josh       │ MNML   │   100,000 │ 2026-04-01 ~            │ active  │
└───────┴──────────┴────────────┴────────┴───────────┴─────────────────────────┴─────────┘
合計: 6件（売上1件 / 仕入5件）

出力例: list --vdu TFHD

$ python -m contract.pipeline list --vdu TFHD
┏━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━┓
┃ ID    ┃ タイプ   ┃ 取引先     ┃ VDU    ┃ 月額(税抜) ┃ 期間                    ┃ ステータス ┃
┡━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━┩
│ C001  │ 売上     │ TFHD       │ TFHD   │ 1,500,000 │ 2026-04-01 ~ 2027-03-31 │ active  │
│ S001  │ 仕入     │ 根崎       │ TFHD   │   500,000 │ 2026-04-01 ~ 2027-03-31 │ active  │
│ S002  │ 仕入     │ 小川哲央   │ TFHD   │   300,000 │ 2026-04-01 ~ 2027-03-31 │ active  │
└───────┴──────────┴────────────┴────────┴───────────┴─────────────────────────┴─────────┘
TFHD VDU: 3件（売上 1,500,000円 / 仕入 800,000円 / 差引 700,000円）

出力例: show C001

$ python -m contract.pipeline show C001
╭──────────────────────────────────────────────╮
│ 契約詳細: C001                               │
├──────────────────────────────────────────────┤
│ タイプ:       売上（クライアント）             │
│ 取引先:       TFHD                            │
│ VDU:          TFHD                            │
│ 契約種別:     コンサルティング                 │
│ 月額:         1,500,000円（税抜）              │
│ 期間:         2026-04-01 ~ 2027-03-31         │
│ 残日数:       360日                            │
│ 自動更新:     なし                             │
│ 支払サイクル: monthly                          │
│ ステータス:   active                           │
│ SharePoint:   /10_Corporate/contracts/clients/ │
│               TFHD/                            │
╰──────────────────────────────────────────────╯

出力例: check

$ python -m contract.pipeline check --days 60
契約期限チェック（警告: 60日以内）
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

⚠ S004 小川さとこ: 残り178日（2026-09-30まで）
  → 更新通知が30日前（2026-09-01）に発行されます

✓ 期限切れの契約はありません
✓ ステータス更新: 0件

連携マップ

利用例

# ops/tfhd/ から
from contract.app.repository import get_monthly_amount

amount = get_monthly_amount("根崎")
# → 500000（円・税抜）

from contract.app.repository import get_active_contracts

tfhd_sales = get_active_contracts(type="sales", vdu="TFHD")
tfhd_purchases = get_active_contracts(type="purchase", vdu="TFHD")

sales_total = sum(c.amount for c in tfhd_sales)
purchase_total = sum(c.amount for c in tfhd_purchases)
margin = sales_total - purchase_total
# → 売上 1,500,000 - 仕入 800,000 = 差引 700,000

# ops/tfhd/app/ に移行期間中のアダプタを提供
from contract.app.repository import get_monthly_amount as _get_amount

def get_target_amount(target_name: str) -> int:
    """targets.json の代わりに contracts.json から金額を取得する。"""
    amount = _get_amount(target_name)
    if amount is None:
        msg = f"契約が見つかりません: {target_name}"
        raise ValueError(msg)
    return amount

フォルダ構成

10_Corporate/
└── contracts/
    ├── clients/                    ← 売上先の契約書
    │   └── TFHD/
    │       └── 2026_コンサルティング契約書.pdf
    └── suppliers/                  ← 委託先の契約書
        ├── 根崎/
        │   └── 2026_業務委託契約書.pdf
        ├── 小川哲央/
        │   └── 2026_業務委託契約書.pdf
        ├── 森屋/
        │   └── 2026_業務委託契約書.pdf
        ├── 小川さとこ/
        │   └── 2026_業務委託契約書.pdf
        └── Josh/
            └── 2026_業務委託契約書.pdf

既存フォルダとの関係

現在の targets.json の役割

移行フェーズ

[
  {"name": "根崎", "email": "...", "client": "TFHD", "contract_id": "S001", ...},
  {"name": "小川", "email": "...", "client": "TFHD", "contract_id": "S002", ...},
  ...
]

入力バリデーション

パフォーマンス

前提条件

• MFクラウドに契約管理APIは存在しない。ローカルJSON管理で開始する
• 契約データの初期投入はCEOが手動で行う（実際の契約書を見て金額・期間を入力）
• pyproject.toml の [tool.setuptools.packages.find] に "contract*" を追加する必要がある
• 追加の外部依存は不要（pydantic, rich は既存依存に含まれる）

制約事項

• contracts.json の手動編集ミスを検出する仕組みがない → check コマンドでバリデーションも行う
• 契約書PDFの自動解析（OCR等）は対象外。手動でJSONに転記する
• MFクラウドとの同期機能は将来拡張（API提供開始を待つ）

トレードオフ