479 lines
17 KiB
Python
479 lines
17 KiB
Python
"""
|
|
e-Gov 電子申請 API v2 クライアント
|
|
OAuth2 + PKCE フローによる認証と電子送達ファイルのダウンロード
|
|
"""
|
|
import hashlib
|
|
import base64
|
|
import secrets
|
|
import os
|
|
import json
|
|
import time
|
|
import re
|
|
import io
|
|
import zipfile
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
# ── エンドポイント定数 ─────────────────────────────────────────
|
|
# EGOV_ENV=sandbox で検証環境、省略または production で本番環境を使用
|
|
_ENV = os.environ.get("EGOV_ENV", "production").lower()
|
|
|
|
if _ENV == "sandbox":
|
|
AUTH_BASE = "https://account2.sbx.e-gov.go.jp/auth"
|
|
API_BASE = "https://api2.sbx.e-gov.go.jp/shinsei/v2"
|
|
else:
|
|
AUTH_BASE = "https://account.e-gov.go.jp/auth"
|
|
API_BASE = "https://api.e-gov.go.jp/shinsei/v2"
|
|
|
|
AUTH_URL = f"{AUTH_BASE}/auth"
|
|
TOKEN_URL = f"{AUTH_BASE}/token"
|
|
LOGOUT_URL = f"{AUTH_BASE}/logout"
|
|
|
|
EGOV_DATA_DIR = Path("/app/egov_data")
|
|
TOKENS_PATH = EGOV_DATA_DIR / "tokens.json"
|
|
CONFIG_PATH = EGOV_DATA_DIR / "api_config.json"
|
|
|
|
|
|
# ── 資格情報 ──────────────────────────────────────────────────
|
|
|
|
def load_api_config() -> dict:
|
|
"""API設定を config ファイルから読み込む（環境変数でも上書き可）"""
|
|
cfg = {}
|
|
if CONFIG_PATH.exists():
|
|
try:
|
|
with open(CONFIG_PATH) as f:
|
|
cfg = json.load(f)
|
|
except Exception:
|
|
pass
|
|
# 環境変数が優先
|
|
client_id = os.environ.get("EGOV_SOFTWARE_ID") or cfg.get("software_id", "")
|
|
client_secret = os.environ.get("EGOV_API_KEY") or cfg.get("api_key", "")
|
|
redirect_uri = os.environ.get("EGOV_REDIRECT_URI") or cfg.get("redirect_uri",
|
|
"http://192.168.0.61:18000/egov.html")
|
|
return {
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
"redirect_uri": redirect_uri,
|
|
}
|
|
|
|
|
|
def save_api_config(software_id: str, api_key: str,
|
|
redirect_uri: str = "http://192.168.0.61:18000/egov.html"):
|
|
"""API設定をファイルに保存"""
|
|
EGOV_DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
with open(CONFIG_PATH, "w") as f:
|
|
json.dump({"software_id": software_id,
|
|
"api_key": api_key,
|
|
"redirect_uri": redirect_uri}, f)
|
|
|
|
|
|
# ── トークン永続化 ────────────────────────────────────────────
|
|
|
|
def load_tokens() -> dict:
|
|
if TOKENS_PATH.exists():
|
|
try:
|
|
with open(TOKENS_PATH) as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
return {}
|
|
|
|
|
|
def save_tokens(tokens: dict):
|
|
EGOV_DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
with open(TOKENS_PATH, "w") as f:
|
|
json.dump(tokens, f)
|
|
|
|
|
|
def delete_tokens():
|
|
if TOKENS_PATH.exists():
|
|
TOKENS_PATH.unlink()
|
|
|
|
|
|
# ── PKCE ─────────────────────────────────────────────────────
|
|
|
|
def generate_pkce() -> tuple[str, str]:
|
|
"""code_verifier と code_challenge (S256) を生成"""
|
|
verifier = secrets.token_urlsafe(48) # 64文字 (43-128の範囲内)
|
|
digest = hashlib.sha256(verifier.encode()).digest()
|
|
challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
|
|
return verifier, challenge
|
|
|
|
|
|
# ── OAuth2 フロー ─────────────────────────────────────────────
|
|
|
|
def build_auth_url(client_id: str, redirect_uri: str,
|
|
challenge: str, state: str) -> str:
|
|
"""OAuth2 認可URL を生成"""
|
|
from urllib.parse import urlencode, quote
|
|
params = {
|
|
"client_id": client_id,
|
|
"response_type": "code",
|
|
"scope": "openid offline_access",
|
|
"redirect_uri": redirect_uri,
|
|
"state": state,
|
|
"code_challenge": challenge,
|
|
"code_challenge_method": "S256",
|
|
}
|
|
# quote_via=quote を使うことでスペースを + ではなく %20 にエンコードする
|
|
return f"{AUTH_URL}?" + urlencode(params, quote_via=quote)
|
|
|
|
|
|
async def exchange_code(code: str, client_id: str, client_secret: str,
|
|
redirect_uri: str, verifier: str) -> dict:
|
|
"""認可コード → アクセストークン・リフレッシュトークン取得"""
|
|
cred = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
TOKEN_URL,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"Authorization": f"Basic {cred}",
|
|
},
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": redirect_uri,
|
|
"code_verifier": verifier,
|
|
},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
async def refresh_tokens_request(refresh_token: str, client_id: str,
|
|
client_secret: str) -> dict:
|
|
"""リフレッシュトークンでアクセストークン再取得"""
|
|
cred = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
TOKEN_URL,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"Authorization": f"Basic {cred}",
|
|
},
|
|
data={
|
|
"grant_type": "refresh_token",
|
|
"refresh_token": refresh_token,
|
|
},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
async def do_logout(refresh_token: str, client_id: str, client_secret: str):
|
|
"""ログアウト（トークン無効化）"""
|
|
cred = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
|
|
async with httpx.AsyncClient() as client:
|
|
try:
|
|
await client.post(
|
|
LOGOUT_URL,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"Authorization": f"Basic {cred}",
|
|
},
|
|
data={"refresh_token": refresh_token},
|
|
timeout=15,
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ── 電子送達 API ──────────────────────────────────────────────
|
|
|
|
async def get_post_list(access_token: str, date_from: str, date_to: str,
|
|
limit: int = 50) -> dict:
|
|
"""
|
|
電子送達一覧取得
|
|
GET /post/lists?date_from=YYYY-MM-DD&date_to=YYYY-MM-DD&limit=N&offset=1
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
f"{API_BASE}/post/lists",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
params={"date_from": date_from, "date_to": date_to,
|
|
"limit": limit, "offset": 1},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
async def get_post_file(access_token: str, post_id: str) -> dict:
|
|
"""
|
|
電子送達取得（通知文書ファイルを base64 で返す）
|
|
GET /post/{post_id}
|
|
Returns: {notice_data_name, file_data (base64), file_name_list}
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
f"{API_BASE}/post/{post_id}",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
timeout=60,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
async def mark_post_done(access_token: str, post_id: str) -> dict:
|
|
"""
|
|
電子送達取得完了登録
|
|
POST /post {"post_id": "..."}
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
f"{API_BASE}/post",
|
|
headers={
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json={"post_id": post_id},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
# ── 申請書作成 API ─────────────────────────────────────────────
|
|
|
|
async def get_procedure_skeleton(access_token: str, proc_id: str) -> dict:
|
|
"""
|
|
手続選択：申請書スケルトン（ひな形）取得
|
|
GET /procedure/{proc_id}
|
|
Returns: {results: {file_data (base64 zip), configuration_file_name, file_info}}
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
f"{API_BASE}/procedure/{proc_id}",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
timeout=60,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
async def submit_apply(access_token: str, proc_id: str,
|
|
file_name: str, file_data_b64: str,
|
|
trial: bool = True) -> dict:
|
|
"""
|
|
申請データ送信
|
|
POST /apply {"proc_id": "...", "send_file": {"file_name": "...", "file_data": "base64..."}}
|
|
Returns: {results: {arrive_id, arrive_date, proc_name, ...}}
|
|
"""
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
if trial:
|
|
headers["X-eGovAPI-Trial"] = "true"
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
f"{API_BASE}/apply",
|
|
headers=headers,
|
|
json={
|
|
"proc_id": proc_id,
|
|
"send_file": {
|
|
"file_name": file_name,
|
|
"file_data": file_data_b64,
|
|
},
|
|
},
|
|
timeout=120,
|
|
)
|
|
if not resp.is_success:
|
|
try:
|
|
err_body = resp.json()
|
|
except Exception:
|
|
err_body = resp.text
|
|
raise RuntimeError(f"HTTP {resp.status_code}: {err_body}")
|
|
return resp.json()
|
|
|
|
|
|
# ── スケルトン XML 修正 ──────────────────────────────────────────
|
|
|
|
def _fill_empty_tag(text: str, tag: str, value: str) -> str:
|
|
"""kousei.xml 内の空タグ <tag/> または <tag></tag> を <tag>value</tag> に置換"""
|
|
# self-closing: <tag/> または <tag />
|
|
text = re.sub(r'<' + re.escape(tag) + r'\s*/>', f'<{tag}>{value}</{tag}>', text)
|
|
# empty: <tag></tag> または <tag> </tag>
|
|
text = re.sub(
|
|
r'<' + re.escape(tag) + r'\s*>\s*</' + re.escape(tag) + r'>',
|
|
f'<{tag}>{value}</{tag}>', text
|
|
)
|
|
return text
|
|
|
|
|
|
def modify_skeleton_for_post_apply(
|
|
file_data_b64: str,
|
|
proc_id: str,
|
|
proc_name: str,
|
|
uketsuke_kikan_id: str,
|
|
shinsei_shubetsu: str = "申込",
|
|
shimei: str = "テスト タロウ",
|
|
shimei_furigana: str = "テスト タロウ",
|
|
yubin: str = "1300022",
|
|
jusho: str = "東京都墨田区江東橋４丁目３１番１０号",
|
|
jigyosho_seiri_mae: str = "41",
|
|
jigyosho_seiri_ato: str = "シスメ",
|
|
jigyosho_bangou: str = "23090",
|
|
) -> str:
|
|
"""
|
|
スケルトン ZIP の kousei.xml に必須フィールドを埋めて再パックする。
|
|
戻り値: 修正済み ZIP の base64 文字列
|
|
"""
|
|
raw = base64.b64decode(file_data_b64)
|
|
buf_out = io.BytesIO()
|
|
|
|
with zipfile.ZipFile(io.BytesIO(raw), 'r') as zin:
|
|
with zipfile.ZipFile(buf_out, 'w', zipfile.ZIP_DEFLATED) as zout:
|
|
for item in zin.infolist():
|
|
data = zin.read(item.filename)
|
|
name_lower = item.filename.lower()
|
|
if name_lower.endswith('/kousei.xml') or name_lower == 'kousei.xml':
|
|
data = _modify_kousei_xml(
|
|
data, proc_id, proc_name, uketsuke_kikan_id,
|
|
shinsei_shubetsu, shimei, shimei_furigana, yubin, jusho
|
|
)
|
|
elif name_lower.endswith('_01.xml') and 'check' not in name_lower:
|
|
# 申請書XML: 事業所情報を埋める
|
|
data = _modify_apply_xml(data, jigyosho_seiri_mae, jigyosho_seiri_ato, jigyosho_bangou)
|
|
zout.writestr(item, data)
|
|
|
|
return base64.b64encode(buf_out.getvalue()).decode()
|
|
|
|
|
|
def _modify_apply_xml(xml_bytes: bytes, seiri_mae: str, seiri_ato: str, bangou: str) -> bytes:
|
|
"""申請書XML の事業所情報を埋める"""
|
|
try:
|
|
text = xml_bytes.decode('utf-8')
|
|
except Exception:
|
|
text = xml_bytes.decode('shift_jis', errors='replace')
|
|
text = _fill_empty_tag(text, '事業所整理記号_前', seiri_mae)
|
|
text = _fill_empty_tag(text, '事業所整理記号_後', seiri_ato)
|
|
text = _fill_empty_tag(text, '事業所番号', bangou)
|
|
return text.encode('utf-8')
|
|
|
|
|
|
def _modify_kousei_xml(
|
|
xml_bytes: bytes,
|
|
proc_id: str,
|
|
proc_name: str,
|
|
uketsuke_kikan_id: str,
|
|
shinsei_shubetsu: str,
|
|
shimei: str,
|
|
shimei_furigana: str,
|
|
yubin: str,
|
|
jusho: str,
|
|
teishutsusaki_id: str = "900API00000000001001001",
|
|
teishutsusaki_name: str = "総務省,行政管理局,API",
|
|
jusho_furigana: str = "トウキョウトスミダクコウトウバシ４チョウメ３１バン１０ゴウ",
|
|
tel: str = "03-1234-5678",
|
|
email: str = "test@example.com",
|
|
) -> bytes:
|
|
"""kousei.xml の必須フィールドをテキスト置換で埋める（名前空間非依存）"""
|
|
try:
|
|
text = xml_bytes.decode('utf-8')
|
|
except Exception:
|
|
text = xml_bytes.decode('shift_jis', errors='replace')
|
|
|
|
fills = [
|
|
('受付行政機関ID', uketsuke_kikan_id),
|
|
('手続ID', proc_id),
|
|
('手続名称', proc_name),
|
|
('申請種別', shinsei_shubetsu),
|
|
('氏名フリガナ', shimei_furigana), # フリガナを先に（氏名より長いので誤置換防止）
|
|
('氏名', shimei),
|
|
('郵便番号', yubin),
|
|
('住所', jusho),
|
|
('提出先識別子', teishutsusaki_id),
|
|
('提出先名称', teishutsusaki_name),
|
|
('住所フリガナ', jusho_furigana),
|
|
('電話番号', tel),
|
|
('電子メールアドレス', email),
|
|
]
|
|
for tag, value in fills:
|
|
text = _fill_empty_tag(text, tag, value)
|
|
|
|
# 申請書属性情報を </提出先情報> の直後に挿入（スケルトンに存在しないため）
|
|
shinsei_attr = (
|
|
'<申請書属性情報>'
|
|
'<申請書様式ID>900A13800001000001</申請書様式ID>'
|
|
'<申請書様式バージョン>0001</申請書様式バージョン>'
|
|
'<申請書様式名称>ＡＰＩテスト用手続（電子送達関係手続）（通）０００１＿０１</申請書様式名称>'
|
|
'<申請書ファイル名称>900A13800001000001_01.xml</申請書ファイル名称>'
|
|
'</申請書属性情報>'
|
|
)
|
|
if '<申請書属性情報>' not in text:
|
|
text = text.replace('</提出先情報>', '</提出先情報>' + shinsei_attr, 1)
|
|
|
|
return text.encode('utf-8')
|
|
|
|
|
|
async def submit_post_apply(access_token: str, proc_id: str,
|
|
file_name: str, file_data_b64: str) -> dict:
|
|
"""
|
|
電子送達利用申込み
|
|
POST /post-apply {"proc_id": "...", "send_file": {"file_name": "...", "file_data": "base64..."}}
|
|
Returns: {results: {arrive_id, arrive_date, proc_name, ...}}
|
|
※電子送達はトライアル非対応のため X-eGovAPI-Trial は送信しない
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
f"{API_BASE}/post-apply",
|
|
headers={
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json={
|
|
"proc_id": proc_id,
|
|
"send_file": {
|
|
"file_name": file_name,
|
|
"file_data": file_data_b64,
|
|
},
|
|
},
|
|
timeout=120,
|
|
)
|
|
if not resp.is_success:
|
|
try:
|
|
err_body = resp.json()
|
|
except Exception:
|
|
err_body = resp.text
|
|
raise RuntimeError(f"HTTP {resp.status_code}: {err_body}")
|
|
return resp.json()
|
|
|
|
|
|
async def get_apply_list(access_token: str, limit: int = 100) -> dict:
|
|
"""
|
|
申請案件一覧取得
|
|
GET /apply/lists
|
|
Returns: {results: {apply_list: [...]}}
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
f"{API_BASE}/apply/lists",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
params={"limit": limit},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
async def get_apply_detail(access_token: str, arrive_id: str) -> dict:
|
|
"""
|
|
申請案件取得（詳細）
|
|
GET /apply/{arrive_id}
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
f"{API_BASE}/apply/{arrive_id}",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|