feat: Odoo APIを使用した請求書発行システムを実装
Co-authored-by: aider (ollama/qwen3-coder:8b) <aider@aider.chat>
This commit is contained in:
joe
commit
fb65e5d381
1 changed files with 313 additions and 0 deletions
313
odoo_invoice_generator.py
Normal file
313
odoo_invoice_generator.py
Normal file
|
|
@ -0,0 +1,313 @@
|
|||
"""
|
||||
Odoo請求書発行システム（REST APIを使用）
|
||||
========================================
|
||||
"""
|
||||
|
||||
import requests
|
||||
from datetime import datetime
|
||||
import json
|
||||
import logging
|
||||
|
||||
|
||||
class OdooAPI:
|
||||
"""
|
||||
Odoo APIクライアント
|
||||
"""
|
||||
|
||||
def __init__(self, base_url: str, db: str, username: str, password: str):
|
||||
self.base_url = f"{base_url}/api/v13"
|
||||
self.db = db
|
||||
self.session = requests.Session()
|
||||
self.login()
|
||||
|
||||
def login(self) -> bool:
|
||||
"""
|
||||
Odoo APIに認証する
|
||||
|
||||
戻り値:
|
||||
bool: 認証成功の場合はTrue、失敗の場合はFalse
|
||||
"""
|
||||
url = f"{self.base_url}/login/db_{self.db}"
|
||||
data = {
|
||||
"jsonrpc": "2.0",
|
||||
"method": "call",
|
||||
"params": {
|
||||
"service": "object",
|
||||
"method": "service_login",
|
||||
"args": [self.db, username, password]
|
||||
}
|
||||
}
|
||||
|
||||
response = self.session.post(url, json=data)
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
self.session_id = result["result"]["session_id"]
|
||||
return True
|
||||
else:
|
||||
logging.error(f"ログイン失敗: {response.text}")
|
||||
return False
|
||||
|
||||
def logout(self) -> bool:
|
||||
"""
|
||||
Odoo APIからログアウトする
|
||||
|
||||
戻り値:
|
||||
bool: ログアウト成功の場合はTrue、失敗の場合はFalse
|
||||
"""
|
||||
url = f"{self.base_url}/login/logout"
|
||||
data = {
|
||||
"jsonrpc": "2.0",
|
||||
"method": "call",
|
||||
"params": {
|
||||
"service": "object",
|
||||
"method": "service_logout",
|
||||
"args": [self.session_id]
|
||||
}
|
||||
}
|
||||
|
||||
response = self.session.post(url, json=data)
|
||||
if response.status_code == 200:
|
||||
return True
|
||||
else:
|
||||
logging.error(f"ログアウト失敗: {response.text}")
|
||||
return False
|
||||
|
||||
def get_partner(self, partner_id: int) -> dict | None:
|
||||
"""
|
||||
顧客情報を取得する
|
||||
|
||||
引数:
|
||||
partner_id (int): 取得する顧客のID
|
||||
|
||||
戻り値:
|
||||
dict | None: 顧客データが見つかった場合、それ以外の場合はNone
|
||||
"""
|
||||
url = f"{self.base_url}/res.partner/{partner_id}"
|
||||
headers = {
|
||||
"Authorization": f"Session {self.session_id}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
response = self.session.get(url, headers=headers)
|
||||
if response.status_code == 200:
|
||||
return response.json()["result"]
|
||||
else:
|
||||
logging.error(f"顧客情報取得失敗: {response.text}")
|
||||
return None
|
||||
|
||||
def get_product(self, product_id: int) -> dict | None:
|
||||
"""
|
||||
商品情報を取得する
|
||||
|
||||
引数:
|
||||
product_id (int): 取得する商品のID
|
||||
|
||||
戻り値:
|
||||
dict | None: 商品データが見つかった場合、それ以外の場合はNone
|
||||
"""
|
||||
url = f"{self.base_url}/product.product/{product_id}"
|
||||
headers = {
|
||||
"Authorization": f"Session {self.session_id}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
response = self.session.get(url, headers=headers)
|
||||
if response.status_code == 200:
|
||||
return response.json()["result"]
|
||||
else:
|
||||
logging.error(f"商品情報取得失敗: {response.text}")
|
||||
return None
|
||||
|
||||
def create_invoice(self, invoice_data: dict) -> dict | None:
|
||||
"""
|
||||
新しい請求書を作成する
|
||||
|
||||
引数:
|
||||
invoice_data (dict): 請求書データ
|
||||
|
||||
戻り値:
|
||||
dict | None: 作成成功した場合の請求書データ、失敗の場合はNone
|
||||
"""
|
||||
url = f"{self.base_url}/account.move"
|
||||
headers = {
|
||||
"Authorization": f"Session {self.session_id}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
response = self.session.post(url, headers=headers, json=invoice_data)
|
||||
if response.status_code == 200:
|
||||
return response.json()["result"]
|
||||
else:
|
||||
logging.error(f"請求書作成失敗: {response.text}")
|
||||
return None
|
||||
|
||||
def get_invoices(self) -> list[dict]:
|
||||
"""
|
||||
全ての請求書を取得する
|
||||
|
||||
戻り値:
|
||||
list[dict]: 請求書リスト
|
||||
"""
|
||||
url = f"{self.base_url}/account.move"
|
||||
headers = {
|
||||
"Authorization": f"Session {self.session_id}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
response = self.session.get(url, headers=headers)
|
||||
if response.status_code == 200:
|
||||
return response.json()["result"]
|
||||
else:
|
||||
logging.error(f"請求書取得失敗: {response.text}")
|
||||
return []
|
||||
|
||||
|
||||
class InvoiceGenerator:
|
||||
"""
|
||||
Odoo請求書生成器
|
||||
"""
|
||||
|
||||
def __init__(self, odoo_api: OdooAPI):
|
||||
self.odoo_api = odoo_api
|
||||
|
||||
def generate_invoice(self, partner_id: int, product_id: int, quantity: int) -> dict | None:
|
||||
"""
|
||||
指定された顧客と商品に対して新しい請求書を生成する
|
||||
|
||||
引数:
|
||||
partner_id (int): 顧客ID
|
||||
product_id (int): 商品ID
|
||||
quantity (int): 売上数量
|
||||
|
||||
戻り値:
|
||||
dict | None: 作成成功した場合の請求書データ、失敗の場合はNone
|
||||
"""
|
||||
|
||||
# 顧客情報を取得
|
||||
partner = self.odoo_api.get_partner(partner_id)
|
||||
if not partner:
|
||||
logging.error(f"顧客 {partner_id} が見つからない")
|
||||
return None
|
||||
|
||||
# 商品情報を取得
|
||||
product = self.odoo_api.get_product(product_id)
|
||||
if not product:
|
||||
logging.error(f"商品 {product_id} が見つからない")
|
||||
return None
|
||||
|
||||
# 合計金額を計算
|
||||
price_unit = product["lst_price"]
|
||||
subtotal = quantity * price_unit
|
||||
tax_rate = 0.08 # 8%の消費税（例、必要に応じて変更可能）
|
||||
tax_amount = subtotal * tax_rate / (1 + tax_rate)
|
||||
total = subtotal + tax_amount
|
||||
|
||||
# 請求書データを作成
|
||||
invoice_data = {
|
||||
"journal_id": 1, # デフォルトの勘定科目ID（必要に応じて変更可能）
|
||||
"partner_id": partner["id"],
|
||||
"date_invoice": datetime.now().strftime("%Y-%m-%d"),
|
||||
"move_type": "out_invoice",
|
||||
"state": "draft", # 初期状態
|
||||
"name": f"{partner['name']} への請求書",
|
||||
"reference": f"INV-{datetime.now().strftime('%Y%m%d%H%M%S')}",
|
||||
"user_id": 1, # デフォルトのユーザーID（必要に応じて変更可能）
|
||||
"company_id": 1, # デフォルトの会社ID（必要に応じて変更可能）
|
||||
|
||||
"invoice_line_ids": [
|
||||
{
|
||||
"product_id": product["id"],
|
||||
"name": f"{product['name']} x{quantity}",
|
||||
"sequence": 10,
|
||||
"type": "line",
|
||||
"quantity": quantity,
|
||||
"price_unit": price_unit,
|
||||
"account_id": product.get("property_account_exp", {}).get("account_id"),
|
||||
"analytic_index_ids": [],
|
||||
"discount": 0.00
|
||||
}
|
||||
],
|
||||
|
||||
# 追加の行、割引などがあればここに追加可能
|
||||
}
|
||||
|
||||
return self.odoo_api.create_invoice(invoice_data)
|
||||
|
||||
def generate_invoices(self, partner_ids: list[int], product_ids: list[int]) -> dict:
|
||||
"""
|
||||
複数の顧客と商品に対して請求書を生成する
|
||||
|
||||
引数:
|
||||
partner_ids (list[int]): 顧客IDリスト
|
||||
product_ids (list[int]): 商品IDリスト
|
||||
|
||||
戻り値:
|
||||
dict: 請求書データとエラーメッセージを含む辞書
|
||||
"""
|
||||
|
||||
# 結果辞書を初期化
|
||||
result = {"invoices": [], "errors": []}
|
||||
|
||||
for partner_id in partner_ids:
|
||||
for product_id in product_ids:
|
||||
try:
|
||||
invoice = self.generate_invoice(partner_id, product_id, 1)
|
||||
if invoice:
|
||||
result["invoices"].append(invoice)
|
||||
except Exception as e:
|
||||
result["errors"].append(f"請求書生成失敗: {str(e)}")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# 実行例
|
||||
if __name__ == "__main__":
|
||||
# Odoo API接続パラメータを設定
|
||||
base_url = "http://localhost:8069"
|
||||
db_name = "mydatabase"
|
||||
username = "admin"
|
||||
password = "password123"
|
||||
|
||||
odoo_api = OdooAPI(base_url, db_name, username, password)
|
||||
|
||||
if not odoo_api.login():
|
||||
print("Odoo APIへの認証失敗")
|
||||
exit(1)
|
||||
|
||||
# インスタンスを作成
|
||||
invoice_generator = InvoiceGenerator(odoo_api)
|
||||
|
||||
# 例1: 単一の請求書を生成
|
||||
partner_id = 2
|
||||
product_id = 3
|
||||
quantity = 5
|
||||
|
||||
invoice = invoice_generator.generate_invoice(partner_id, product_id, quantity)
|
||||
if invoice:
|
||||
print("請求書作成成功:")
|
||||
print(json.dumps(invoice, indent=4))
|
||||
else:
|
||||
print(f"顧客 {partner_id} と商品 {product_id} に対する請求書作成失敗")
|
||||
|
||||
# 例2: 複数の請求書を生成
|
||||
partner_ids = [1, 2, 3]
|
||||
product_ids = [101, 102, 103]
|
||||
|
||||
result = invoice_generator.generate_invoices(partner_ids, product_ids)
|
||||
print("\n請求書生成結果:")
|
||||
print(f"作成された請求書: {len(result['invoices'])}")
|
||||
print(f"エラー: {len(result['errors'])}")
|
||||
|
||||
# 請求書を表示
|
||||
for i, invoice in enumerate(result["invoices"]):
|
||||
print(f"\n請求書 {i+1}:")
|
||||
print(json.dumps(invoice, indent=4))
|
||||
|
||||
# エラーを表示
|
||||
if result["errors"]:
|
||||
print("\nエラーログ:")
|
||||
for error in result["errors"]:
|
||||
print(error)
|
||||
|
||||
# Odoo APIからログアウト
|
||||
odoo_api.logout()
|
||||
Loading…
Reference in a new issue