1122 lines
39 KiB
Python
Executable file
1122 lines
39 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""
|
||
اسکریپت انتقال فاکتورهای فروش، خرید، برگشت از فروش و برگشت از خرید از hesabixOld به hesabixpy
|
||
|
||
این اسکریپت:
|
||
- فاکتورهای sell, buy, rfsell, rfbuy را منتقل می‌کند
|
||
- از سرویس invoice_service.create_invoice() استفاده می‌کند
|
||
- ثبت‌های حسابداری به صورت خودکار انجام می‌شود
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import argparse
|
||
from typing import List, Dict, Any, Optional, Tuple
|
||
from datetime import datetime, date
|
||
from decimal import Decimal, InvalidOperation
|
||
import re
|
||
|
||
# اضافه کردن مسیر پروژه به sys.path
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from sqlalchemy import create_engine, text
|
||
from sqlalchemy.orm import sessionmaker, Session
|
||
|
||
|
||
def convert_timestamp_to_datetime(timestamp_str: str | None) -> datetime | None:
|
||
"""تبدیل timestamp string به datetime"""
|
||
if not timestamp_str:
|
||
return None
|
||
|
||
try:
|
||
if timestamp_str.strip().isdigit():
|
||
return datetime.fromtimestamp(int(timestamp_str))
|
||
except (ValueError, TypeError, OSError):
|
||
pass
|
||
|
||
return None
|
||
|
||
|
||
def convert_persian_date_to_date(date_str: str | None) -> date | None:
|
||
"""تبدیل تاریخ شمسی string به date"""
|
||
if not date_str or not date_str.strip():
|
||
return None
|
||
|
||
date_str = date_str.strip()
|
||
|
||
# فرمت: 1403/10/05
|
||
match = re.match(r'^(\d{4})/(\d{1,2})/(\d{1,2})$', date_str)
|
||
if match:
|
||
year, month, day = match.groups()
|
||
try:
|
||
import jdatetime
|
||
jd = jdatetime.date(int(year), int(month), int(day))
|
||
gd = jd.togregorian()
|
||
return gd
|
||
except ImportError:
|
||
# تبدیل تقریبی
|
||
gregorian_year = int(year) + 621
|
||
try:
|
||
return date(gregorian_year, int(month), int(day))
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
return None
|
||
|
||
|
||
def convert_amount(amount_str: str | None) -> Decimal:
|
||
"""تبدیل amount از string به decimal"""
|
||
if not amount_str:
|
||
return Decimal(0)
|
||
|
||
try:
|
||
cleaned = str(amount_str).strip().replace(',', '').replace(' ', '').replace('،', '')
|
||
if not cleaned or cleaned == '0' or cleaned == '':
|
||
return Decimal(0)
|
||
return Decimal(cleaned)
|
||
except (ValueError, InvalidOperation, TypeError):
|
||
return Decimal(0)
|
||
|
||
|
||
class InvoiceMigration:
|
||
def __init__(self, old_db_name: str = "hesabixOld", new_db_name: str = "hesabixpy",
|
||
db_user: str = "root", db_password: str = "136431",
|
||
db_host: str = "localhost", db_port: int = 3306):
|
||
"""ایجاد اتصال به هر دو دیتابیس"""
|
||
self.old_db_name = old_db_name
|
||
self.new_db_name = new_db_name
|
||
|
||
# اتصال به دیتابیس قدیمی
|
||
old_dsn = f"mysql+pymysql://{db_user}:{db_password}@{db_host}:{db_port}/{old_db_name}"
|
||
old_engine = create_engine(
|
||
old_dsn,
|
||
echo=False,
|
||
pool_pre_ping=True,
|
||
pool_size=10,
|
||
max_overflow=20,
|
||
pool_recycle=3600,
|
||
connect_args={
|
||
"connect_timeout": 60,
|
||
"read_timeout": 300,
|
||
"write_timeout": 300,
|
||
"charset": "utf8mb4",
|
||
"init_command": "SET sql_mode='STRICT_TRANS_TABLES,NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO'"
|
||
}
|
||
)
|
||
self.old_db = sessionmaker(bind=old_engine)()
|
||
|
||
# اتصال به دیتابیس جدید
|
||
new_dsn = f"mysql+pymysql://{db_user}:{db_password}@{db_host}:{db_port}/{new_db_name}"
|
||
new_engine = create_engine(
|
||
new_dsn,
|
||
echo=False,
|
||
pool_pre_ping=True,
|
||
pool_size=10,
|
||
max_overflow=20,
|
||
pool_recycle=3600,
|
||
connect_args={
|
||
"connect_timeout": 60,
|
||
"read_timeout": 300,
|
||
"write_timeout": 300,
|
||
"charset": "utf8mb4",
|
||
"init_command": "SET sql_mode='STRICT_TRANS_TABLES,NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO'"
|
||
}
|
||
)
|
||
self.new_db = sessionmaker(bind=new_engine)()
|
||
|
||
# Cache ها برای بهینه‌سازی
|
||
self.default_currency_cache: Dict[int, int] = {} # {business_id: currency_id}
|
||
self.default_fiscal_year_cache: Dict[int, int] = {} # {business_id: fiscal_year_id}
|
||
self.migrated_old_doc_ids: set[int] = set() # Cache برای migrated documents
|
||
self.migrated_codes_cache: Dict[Tuple[int, str], bool] = {} # {(business_id, code): True}
|
||
|
||
# آمار
|
||
self.stats = {
|
||
"invoices_processed": 0,
|
||
"invoices_migrated": 0,
|
||
"invoices_skipped": 0,
|
||
"errors": 0,
|
||
"error_details": []
|
||
}
|
||
|
||
def create_business_id_mapping(self) -> Dict[int, int]:
|
||
"""ایجاد mapping بین business_id قدیمی و جدید با case-insensitive"""
|
||
query = text("""
|
||
SELECT
|
||
old_business.id as old_business_id,
|
||
new_business.id as new_business_id
|
||
FROM hesabixOld.business old_business
|
||
INNER JOIN hesabixOld.user old_user ON old_business.owner_id = old_user.id
|
||
INNER JOIN hesabixpy.users new_user ON (
|
||
(old_user.email IS NOT NULL AND new_user.email IS NOT NULL AND LOWER(TRIM(old_user.email)) = LOWER(TRIM(new_user.email))) OR
|
||
(old_user.mobile IS NOT NULL AND new_user.mobile IS NOT NULL AND REPLACE(REPLACE(REPLACE(old_user.mobile, '+', ''), ' ', ''), '-', '') = REPLACE(REPLACE(REPLACE(new_user.mobile, '+', ''), ' ', ''), '-', ''))
|
||
)
|
||
INNER JOIN hesabixpy.businesses new_business ON (
|
||
new_business.owner_id = new_user.id
|
||
AND new_business.name = old_business.name
|
||
)
|
||
WHERE old_user.active = 1
|
||
""")
|
||
|
||
results = self.old_db.execute(query).fetchall()
|
||
mapping = {}
|
||
for row in results:
|
||
mapping[row.old_business_id] = row.new_business_id
|
||
|
||
print(f"✅ Business ID mapping ایجاد شد: {len(mapping)} کسب و کار")
|
||
return mapping
|
||
|
||
def create_user_id_mapping(self) -> Dict[int, int]:
|
||
"""ایجاد mapping بین user_id قدیمی و جدید با case-insensitive"""
|
||
query = text("""
|
||
SELECT
|
||
old.id as old_user_id,
|
||
new.id as new_user_id
|
||
FROM hesabixOld.user old
|
||
INNER JOIN hesabixpy.users new ON (
|
||
(old.email IS NOT NULL AND new.email IS NOT NULL AND LOWER(TRIM(old.email)) = LOWER(TRIM(new.email))) OR
|
||
(old.mobile IS NOT NULL AND new.mobile IS NOT NULL AND REPLACE(REPLACE(REPLACE(old.mobile, '+', ''), ' ', ''), '-', '') = REPLACE(REPLACE(REPLACE(new.mobile, '+', ''), ' ', ''), '-', ''))
|
||
)
|
||
WHERE old.active = 1
|
||
""")
|
||
|
||
results = self.old_db.execute(query).fetchall()
|
||
mapping = {}
|
||
for row in results:
|
||
mapping[row.old_user_id] = row.new_user_id
|
||
|
||
print(f"✅ User ID mapping ایجاد شد: {len(mapping)} کاربر")
|
||
return mapping
|
||
|
||
def create_fiscal_year_mapping(self, business_id_mapping: Dict[int, int]) -> Dict[Tuple[int, int], int]:
|
||
"""ایجاد mapping بین fiscal_year_id قدیمی و جدید"""
|
||
# mapping: (old_business_id, old_year_id) -> new_year_id
|
||
mapping = {}
|
||
|
||
for old_bid, new_bid in business_id_mapping.items():
|
||
# در جدول قدیمی start و end به صورت timestamp (varchar) هستند
|
||
# باید آنها را به date تبدیل کنیم
|
||
query = text("""
|
||
SELECT
|
||
old.id as old_year_id,
|
||
new.id as new_year_id
|
||
FROM {}.year old
|
||
INNER JOIN {}.fiscal_years new ON (
|
||
new.business_id = :new_bid
|
||
AND new.start_date = DATE(FROM_UNIXTIME(CAST(old.start AS UNSIGNED)))
|
||
AND new.end_date = DATE(FROM_UNIXTIME(CAST(old.end AS UNSIGNED)))
|
||
)
|
||
WHERE old.bid_id = :old_bid
|
||
""".format(self.old_db_name, self.new_db_name))
|
||
|
||
try:
|
||
results = self.old_db.execute(query, {
|
||
"old_bid": old_bid,
|
||
"new_bid": new_bid
|
||
}).fetchall()
|
||
|
||
for row in results:
|
||
mapping[(old_bid, row.old_year_id)] = row.new_year_id
|
||
except Exception as e:
|
||
# اگر تبدیل تاریخ با مشکل مواجه شد، فقط بر اساس business_id و label (یا label) جستجو کنیم
|
||
query2 = text("""
|
||
SELECT
|
||
old.id as old_year_id,
|
||
new.id as new_year_id
|
||
FROM {}.year old
|
||
INNER JOIN {}.fiscal_years new ON (
|
||
new.business_id = :new_bid
|
||
AND new.title = old.label
|
||
)
|
||
WHERE old.bid_id = :old_bid
|
||
LIMIT 1
|
||
""".format(self.old_db_name, self.new_db_name))
|
||
|
||
try:
|
||
results = self.old_db.execute(query2, {
|
||
"old_bid": old_bid,
|
||
"new_bid": new_bid
|
||
}).fetchall()
|
||
|
||
for row in results:
|
||
mapping[(old_bid, row.old_year_id)] = row.new_year_id
|
||
except Exception:
|
||
pass # اگر پیدا نشد، skip می‌کنیم
|
||
|
||
print(f"✅ Fiscal Year mapping ایجاد شد: {len(mapping)} سال مالی")
|
||
return mapping
|
||
|
||
def create_currency_mapping(self) -> Dict[int, int]:
|
||
"""ایجاد mapping بین currency_id قدیمی و جدید"""
|
||
query = text("""
|
||
SELECT
|
||
old.id as old_currency_id,
|
||
new.id as new_currency_id
|
||
FROM {}.money old
|
||
INNER JOIN {}.currencies new ON (
|
||
BINARY new.code = BINARY old.name
|
||
AND BINARY new.symbol = BINARY old.symbol
|
||
)
|
||
""".format(self.old_db_name, self.new_db_name))
|
||
|
||
results = self.old_db.execute(query).fetchall()
|
||
mapping = {}
|
||
for row in results:
|
||
mapping[row.old_currency_id] = row.new_currency_id
|
||
|
||
print(f"✅ Currency mapping ایجاد شد: {len(mapping)} ارز")
|
||
return mapping
|
||
|
||
def create_person_id_mapping(self, business_id_mapping: Dict[int, int]) -> Dict[Tuple[int, int], int]:
|
||
"""ایجاد mapping بین person_id قدیمی و جدید"""
|
||
# mapping: (old_business_id, old_person_id) -> new_person_id
|
||
mapping = {}
|
||
|
||
for old_bid, new_bid in business_id_mapping.items():
|
||
query = text("""
|
||
SELECT
|
||
old.id as old_person_id,
|
||
new.id as new_person_id
|
||
FROM {}.person old
|
||
INNER JOIN {}.persons new ON (
|
||
new.business_id = :new_bid
|
||
AND BINARY new.code = BINARY old.code
|
||
)
|
||
WHERE old.bid_id = :old_bid
|
||
""".format(self.old_db_name, self.new_db_name))
|
||
|
||
results = self.old_db.execute(query, {
|
||
"old_bid": old_bid,
|
||
"new_bid": new_bid
|
||
}).fetchall()
|
||
|
||
for row in results:
|
||
mapping[(old_bid, row.old_person_id)] = row.new_person_id
|
||
|
||
print(f"✅ Person ID mapping ایجاد شد: {len(mapping)} شخص")
|
||
return mapping
|
||
|
||
def create_product_id_mapping(self, business_id_mapping: Dict[int, int]) -> Dict[Tuple[int, int], int]:
|
||
"""ایجاد mapping بین product_id قدیمی و جدید"""
|
||
# mapping: (old_business_id, old_commodity_id) -> new_product_id
|
||
mapping = {}
|
||
|
||
for old_bid, new_bid in business_id_mapping.items():
|
||
query = text("""
|
||
SELECT
|
||
old.id as old_commodity_id,
|
||
new.id as new_product_id
|
||
FROM {}.commodity old
|
||
INNER JOIN {}.products new ON (
|
||
new.business_id = :new_bid
|
||
AND BINARY new.code = BINARY old.code
|
||
)
|
||
WHERE old.bid_id = :old_bid
|
||
""".format(self.old_db_name, self.new_db_name))
|
||
|
||
results = self.old_db.execute(query, {
|
||
"old_bid": old_bid,
|
||
"new_bid": new_bid
|
||
}).fetchall()
|
||
|
||
for row in results:
|
||
mapping[(old_bid, row.old_commodity_id)] = row.new_product_id
|
||
|
||
print(f"✅ Product ID mapping ایجاد شد: {len(mapping)} کالا")
|
||
return mapping
|
||
|
||
def ensure_policy_exists(self, business_id: int):
|
||
"""اطمینان از وجود سیاست درآمدزایی برای کسب‌وکار"""
|
||
# بررسی وجود سیاست
|
||
query = text("""
|
||
SELECT COUNT(*) FROM document_usage_policies
|
||
WHERE business_id = :business_id AND is_active = 1
|
||
""")
|
||
result = self.new_db.execute(query, {"business_id": business_id}).scalar()
|
||
|
||
if result == 0:
|
||
# ایجاد سیاست free
|
||
query_insert = text("""
|
||
INSERT INTO document_usage_policies
|
||
(business_id, policy_type, title, priority, is_active, config, created_at, updated_at)
|
||
VALUES
|
||
(:business_id, 'free', 'سیاست رایگان برای انتقال', 100, 1, NULL, NOW(), NOW())
|
||
""")
|
||
self.new_db.execute(query_insert, {"business_id": business_id})
|
||
self.new_db.commit()
|
||
|
||
def get_old_invoices(self, invoice_type: str, start_id: Optional[int] = None,
|
||
limit: Optional[int] = None, offset: Optional[int] = None,
|
||
business_ids: Optional[List[int]] = None) -> List[Dict[str, Any]]:
|
||
"""دریافت فاکتورها از دیتابیس قدیمی"""
|
||
query_str = """
|
||
SELECT
|
||
d.id, d.bid_id, d.submitter_id, d.year_id, d.money_id,
|
||
d.date_submit, d.date, d.type, d.code, d.des, d.amount,
|
||
d.is_preview, d.is_approved, d.project_id, d.tax_percent,
|
||
d.discount_type, d.discount_percent
|
||
FROM {}.hesabdari_doc d
|
||
WHERE d.type = :invoice_type
|
||
""".format(self.old_db_name)
|
||
|
||
params = {"invoice_type": invoice_type}
|
||
|
||
if business_ids:
|
||
placeholders = ','.join([f':bid_{i}' for i in range(len(business_ids))])
|
||
query_str += f" AND d.bid_id IN ({placeholders})"
|
||
for i, bid in enumerate(business_ids):
|
||
params[f'bid_{i}'] = bid
|
||
elif start_id:
|
||
query_str += " AND d.id >= :start_id"
|
||
params["start_id"] = start_id
|
||
|
||
# فقط اسناد تایید شده
|
||
query_str += " AND d.is_approved = 1"
|
||
|
||
query_str += " ORDER BY d.bid_id, d.id ASC"
|
||
|
||
if limit:
|
||
query_str += " LIMIT :limit"
|
||
params["limit"] = limit
|
||
# فقط زمانی از offset استفاده می‌کنیم که start_id استفاده نشده باشد
|
||
if offset and offset > 0 and not start_id:
|
||
query_str += " OFFSET :offset"
|
||
params["offset"] = offset
|
||
|
||
results = self.old_db.execute(text(query_str), params).fetchall()
|
||
|
||
invoices = []
|
||
for row in results:
|
||
invoices.append({
|
||
"id": row[0],
|
||
"business_id": row[1],
|
||
"submitter_id": row[2],
|
||
"year_id": row[3],
|
||
"currency_id": row[4],
|
||
"date_submit": row[5],
|
||
"date": row[6],
|
||
"type": row[7],
|
||
"code": row[8],
|
||
"description": row[9],
|
||
"amount": row[10],
|
||
"is_preview": row[11],
|
||
"is_approved": row[12],
|
||
"project_id": row[13],
|
||
"tax_percent": row[14],
|
||
"discount_type": row[15],
|
||
"discount_percent": row[16]
|
||
})
|
||
|
||
return invoices
|
||
|
||
def get_invoice_rows(self, doc_id: int) -> List[Dict[str, Any]]:
|
||
"""دریافت سطرهای فاکتور از دیتابیس قدیمی"""
|
||
query = text("""
|
||
SELECT
|
||
r.id, r.ref_id, r.person_id, r.commodity_id, r.bs, r.bd, r.des,
|
||
r.commdity_count, r.discount, r.tax, r.bank_id, r.cashdesk_id,
|
||
r.salary_id, r.cheque_id, t.name as account_name, t.type as account_type
|
||
FROM {}.hesabdari_row r
|
||
LEFT JOIN {}.hesabdari_table t ON r.ref_id = t.id
|
||
WHERE r.doc_id = :doc_id
|
||
ORDER BY r.id ASC
|
||
""".format(self.old_db_name, self.old_db_name))
|
||
|
||
results = self.old_db.execute(query, {"doc_id": doc_id}).fetchall()
|
||
|
||
def get_invoice_rows_bulk(self, doc_ids: List[int]) -> Dict[int, List[Dict[str, Any]]]:
|
||
"""دریافت سطرهای چند فاکتور به صورت bulk"""
|
||
if not doc_ids:
|
||
return {}
|
||
|
||
placeholders = ','.join([f':doc_id_{i}' for i in range(len(doc_ids))])
|
||
query = text(f"""
|
||
SELECT
|
||
r.doc_id, r.id, r.ref_id, r.person_id, r.commodity_id, r.bs, r.bd, r.des,
|
||
r.commdity_count, r.discount, r.tax, r.bank_id, r.cashdesk_id,
|
||
r.salary_id, r.cheque_id, t.name as account_name, t.type as account_type
|
||
FROM {self.old_db_name}.hesabdari_row r
|
||
LEFT JOIN {self.old_db_name}.hesabdari_table t ON r.ref_id = t.id
|
||
WHERE r.doc_id IN ({placeholders})
|
||
ORDER BY r.doc_id, r.id ASC
|
||
""")
|
||
|
||
params = {f'doc_id_{i}': doc_id for i, doc_id in enumerate(doc_ids)}
|
||
results = self.old_db.execute(query, params).fetchall()
|
||
|
||
rows_dict: Dict[int, List[Dict[str, Any]]] = {}
|
||
for row in results:
|
||
doc_id = row[0]
|
||
if doc_id not in rows_dict:
|
||
rows_dict[doc_id] = []
|
||
|
||
rows_dict[doc_id].append({
|
||
"id": row[1],
|
||
"ref_id": row[2],
|
||
"person_id": row[3],
|
||
"commodity_id": row[4],
|
||
"debit": row[5],
|
||
"credit": row[6],
|
||
"description": row[7],
|
||
"quantity": row[8],
|
||
"discount": row[9],
|
||
"tax": row[10],
|
||
"bank_id": row[11],
|
||
"cashdesk_id": row[12],
|
||
"salary_id": row[13],
|
||
"cheque_id": row[14],
|
||
"account_name": row[15],
|
||
"account_type": row[16]
|
||
})
|
||
|
||
return rows_dict
|
||
|
||
def _load_migrated_documents_cache(self, business_id_mapping: Dict[int, int]):
|
||
"""بارگذاری cache اسناد منتقل شده از دیتابیس"""
|
||
if not business_id_mapping:
|
||
return
|
||
|
||
new_business_ids = list(business_id_mapping.values())
|
||
if not new_business_ids:
|
||
return
|
||
|
||
placeholders = ','.join([f':bid_{i}' for i in range(len(new_business_ids))])
|
||
|
||
# Load migrated old_document_ids
|
||
query1 = text(f"""
|
||
SELECT JSON_EXTRACT(extra_info, '$.old_document_id') as old_doc_id
|
||
FROM documents
|
||
WHERE business_id IN ({placeholders})
|
||
AND extra_info LIKE '%migration%'
|
||
AND JSON_EXTRACT(extra_info, '$.old_document_id') IS NOT NULL
|
||
""")
|
||
params1 = {f'bid_{i}': bid for i, bid in enumerate(new_business_ids)}
|
||
results1 = self.new_db.execute(query1, params1).fetchall()
|
||
|
||
for row in results1:
|
||
try:
|
||
old_doc_id = int(row[0])
|
||
self.migrated_old_doc_ids.add(old_doc_id)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# Load migrated codes
|
||
query2 = text(f"""
|
||
SELECT business_id, code
|
||
FROM documents
|
||
WHERE business_id IN ({placeholders})
|
||
AND code IS NOT NULL
|
||
""")
|
||
results2 = self.new_db.execute(query2, params1).fetchall()
|
||
|
||
for row in results2:
|
||
business_id = row[0]
|
||
code = row[1]
|
||
if business_id and code:
|
||
self.migrated_codes_cache[(business_id, code)] = True
|
||
|
||
print(f" 📋 Cache بارگذاری شد: {len(self.migrated_old_doc_ids)} سند منتقل شده، {len(self.migrated_codes_cache)} کد")
|
||
|
||
def _batch_duplicate_check(self, invoices: List[Dict[str, Any]], business_id_mapping: Dict[int, int]):
|
||
"""بررسی تکراری بودن فاکتورها به صورت batch"""
|
||
if not invoices:
|
||
return
|
||
|
||
# جمع‌آوری business_id و code ها
|
||
business_codes: Dict[int, List[str]] = {}
|
||
for invoice in invoices:
|
||
old_bid = invoice.get('business_id')
|
||
new_bid = business_id_mapping.get(old_bid)
|
||
code = invoice.get('code')
|
||
|
||
if new_bid and code:
|
||
if new_bid not in business_codes:
|
||
business_codes[new_bid] = []
|
||
business_codes[new_bid].append(code)
|
||
|
||
# Query batch برای یافتن کدهای تکراری
|
||
for new_bid, codes in business_codes.items():
|
||
if not codes:
|
||
continue
|
||
|
||
placeholders = ','.join([f':code_{i}' for i in range(len(codes))])
|
||
query = text(f"""
|
||
SELECT code
|
||
FROM documents
|
||
WHERE business_id = :business_id
|
||
AND BINARY code IN ({placeholders})
|
||
""")
|
||
params = {"business_id": new_bid}
|
||
for i, code in enumerate(codes):
|
||
params[f'code_{i}'] = code
|
||
|
||
results = self.new_db.execute(query, params).fetchall()
|
||
|
||
# اضافه کردن به cache
|
||
for row in results:
|
||
if row[0]:
|
||
self.migrated_codes_cache[(new_bid, row[0])] = True
|
||
|
||
def extract_invoice_lines(self, rows: List[Dict[str, Any]], invoice_type: str,
|
||
business_id_mapping: Dict[int, int],
|
||
product_mapping: Dict[Tuple[int, int], int]) -> List[Dict[str, Any]]:
|
||
"""استخراج و تبدیل سطرهای فاکتور به فرمت جدید"""
|
||
lines = []
|
||
|
||
# شناسایی ref_id برای کالاها (معمولاً 59 برای فروش کالا)
|
||
# و ref_id برای شخص (معمولاً 3 برای حساب‌های دریافتی)
|
||
|
||
for row in rows:
|
||
ref_id = row.get("ref_id")
|
||
commodity_id = row.get("commodity_id")
|
||
person_id = row.get("person_id")
|
||
|
||
# اگر commodity_id وجود دارد، این یک سطر کالا است
|
||
if commodity_id:
|
||
# پیدا کردن business_id قدیمی از rows (از اولین row می‌توانیم استفاده کنیم)
|
||
# یا باید از invoice استفاده کنیم - در اینجا فرض می‌کنیم که در invoice موجود است
|
||
|
||
# محاسبه unit_price
|
||
debit = convert_amount(row.get("debit"))
|
||
credit = convert_amount(row.get("credit"))
|
||
quantity = convert_amount(row.get("quantity") or 1)
|
||
|
||
if quantity == 0:
|
||
quantity = Decimal(1)
|
||
|
||
# unit_price = (بدهکار یا بستانکار) / quantity
|
||
unit_price = (debit + credit) / quantity if quantity > 0 else Decimal(0)
|
||
|
||
# تخفیف
|
||
discount_amount = convert_amount(row.get("discount") or 0)
|
||
line_discount = discount_amount
|
||
|
||
# مالیات
|
||
tax_amount = convert_amount(row.get("tax") or 0)
|
||
|
||
# محاسبه tax_percent
|
||
taxable_amount = (quantity * unit_price) - line_discount
|
||
tax_percent = Decimal(0)
|
||
if taxable_amount > 0 and tax_amount > 0:
|
||
tax_percent = (tax_amount / taxable_amount) * 100
|
||
|
||
line = {
|
||
"product_id": None, # باید بعداً set شود
|
||
"quantity": float(quantity),
|
||
"extra_info": {
|
||
"unit_price": float(unit_price),
|
||
"line_discount": float(line_discount),
|
||
"tax_amount": float(tax_amount),
|
||
"tax_rate": float(tax_percent)
|
||
},
|
||
"description": row.get("description"),
|
||
"_old_commodity_id": commodity_id, # برای mapping بعدی
|
||
"_old_business_id": None # باید set شود
|
||
}
|
||
|
||
lines.append(line)
|
||
|
||
return lines
|
||
|
||
def migrate_invoice(self, old_invoice: Dict[str, Any],
|
||
business_id_mapping: Dict[int, int],
|
||
user_id_mapping: Dict[int, int],
|
||
fiscal_year_mapping: Dict[Tuple[int, int], int],
|
||
currency_mapping: Dict[int, int],
|
||
person_mapping: Dict[Tuple[int, int], int],
|
||
product_mapping: Dict[Tuple[int, int], int],
|
||
dry_run: bool = False) -> Optional[int]:
|
||
"""انتقال یک فاکتور"""
|
||
try:
|
||
# تعریف متغیرهای اولیه
|
||
old_code = old_invoice.get('code')
|
||
old_document_id = old_invoice.get('id')
|
||
|
||
# نگاشت شناسه‌ها
|
||
old_business_id = old_invoice.get('business_id')
|
||
new_business_id = business_id_mapping.get(old_business_id)
|
||
|
||
if not new_business_id:
|
||
self.stats["invoices_skipped"] += 1
|
||
return None
|
||
|
||
# بررسی اینکه آیا این سند قبلاً منتقل شده یا نه (با cache)
|
||
if old_document_id in self.migrated_old_doc_ids:
|
||
# سند قبلاً منتقل شده، skip می‌کنیم
|
||
return None
|
||
|
||
old_user_id = old_invoice.get('submitter_id')
|
||
new_user_id = user_id_mapping.get(old_user_id)
|
||
|
||
if not new_user_id:
|
||
self.stats["invoices_skipped"] += 1
|
||
return None
|
||
|
||
old_currency_id = old_invoice.get('currency_id')
|
||
new_currency_id = currency_mapping.get(old_currency_id)
|
||
|
||
# اگر ارز پیدا نشد، از ارز پیش‌فرض کسب‌وکار استفاده می‌کنیم (با cache)
|
||
if not new_currency_id:
|
||
if new_business_id in self.default_currency_cache:
|
||
new_currency_id = self.default_currency_cache[new_business_id]
|
||
else:
|
||
query = text("""
|
||
SELECT default_currency_id FROM businesses
|
||
WHERE id = :business_id
|
||
""")
|
||
result = self.new_db.execute(query, {"business_id": new_business_id}).first()
|
||
if result and result[0]:
|
||
new_currency_id = result[0]
|
||
self.default_currency_cache[new_business_id] = new_currency_id
|
||
else:
|
||
if self.stats["invoices_skipped"] < 5:
|
||
print(f" ⚠️ فاکتور {old_invoice.get('code')} رد شد: ارز پیدا نشد و ارز پیش‌فرض کسب‌وکار هم نیست")
|
||
self.stats["invoices_skipped"] += 1
|
||
return None
|
||
|
||
old_year_id = old_invoice.get('year_id')
|
||
fiscal_year_id = fiscal_year_mapping.get((old_business_id, old_year_id))
|
||
|
||
# اگر سال مالی پیدا نشد، از سال مالی پیش‌فرض (جاری) کسب‌وکار استفاده می‌کنیم (با cache)
|
||
if not fiscal_year_id:
|
||
if new_business_id in self.default_fiscal_year_cache:
|
||
fiscal_year_id = self.default_fiscal_year_cache[new_business_id]
|
||
else:
|
||
# ابتدا سعی می‌کنیم سال مالی جاری (is_last = True) را پیدا کنیم
|
||
query = text("""
|
||
SELECT id FROM fiscal_years
|
||
WHERE business_id = :business_id AND is_last = 1
|
||
LIMIT 1
|
||
""")
|
||
result = self.new_db.execute(query, {"business_id": new_business_id}).first()
|
||
if result:
|
||
fiscal_year_id = result[0]
|
||
self.default_fiscal_year_cache[new_business_id] = fiscal_year_id
|
||
else:
|
||
# اگر سال مالی جاری پیدا نشد، آخرین سال مالی را برمی‌گردانیم
|
||
query2 = text("""
|
||
SELECT id FROM fiscal_years
|
||
WHERE business_id = :business_id
|
||
ORDER BY start_date DESC
|
||
LIMIT 1
|
||
""")
|
||
result2 = self.new_db.execute(query2, {"business_id": new_business_id}).first()
|
||
if result2:
|
||
fiscal_year_id = result2[0]
|
||
self.default_fiscal_year_cache[new_business_id] = fiscal_year_id
|
||
else:
|
||
if self.stats["invoices_skipped"] < 5:
|
||
print(f" ⚠️ فاکتور {old_invoice.get('code')} رد شد: سال مالی پیدا نشد (old_year_id: {old_year_id})")
|
||
self.stats["invoices_skipped"] += 1
|
||
return None
|
||
|
||
# تبدیل تاریخ
|
||
doc_date = convert_persian_date_to_date(old_invoice.get('date'))
|
||
if not doc_date:
|
||
date_submit = convert_timestamp_to_datetime(old_invoice.get('date_submit'))
|
||
if date_submit:
|
||
doc_date = date_submit.date()
|
||
else:
|
||
doc_date = date.today()
|
||
|
||
# تشخیص نوع فاکتور
|
||
old_type = old_invoice.get('type')
|
||
type_mapping = {
|
||
"sell": "invoice_sales",
|
||
"buy": "invoice_purchase",
|
||
"rfsell": "invoice_sales_return",
|
||
"rfbuy": "invoice_purchase_return"
|
||
}
|
||
new_invoice_type = type_mapping.get(old_type)
|
||
|
||
if not new_invoice_type:
|
||
self.stats["invoices_skipped"] += 1
|
||
return None
|
||
|
||
# دریافت سطرهای فاکتور (استفاده از cache اگر موجود باشد)
|
||
if '_cached_rows' in old_invoice:
|
||
old_rows = old_invoice['_cached_rows']
|
||
else:
|
||
old_rows = self.get_invoice_rows(old_invoice.get('id'))
|
||
|
||
if not old_rows:
|
||
self.stats["invoices_skipped"] += 1
|
||
return None
|
||
|
||
# استخراج person_id از سطرها
|
||
person_id = None
|
||
person_ids_found = []
|
||
for row in old_rows:
|
||
row_person_id = row.get('person_id')
|
||
if row_person_id is not None:
|
||
try:
|
||
row_person_id_int = int(row_person_id)
|
||
person_ids_found.append(row_person_id_int)
|
||
mapped_person_id = person_mapping.get((old_business_id, row_person_id_int))
|
||
if mapped_person_id:
|
||
person_id = mapped_person_id
|
||
break
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# اگر person_id پیدا نشد، لاگ می‌کنیم
|
||
if not person_id and person_ids_found:
|
||
if self.stats["errors"] < 3:
|
||
print(f" ⚠️ فاکتور {old_code}: person_id در mapping پیدا نشد (old_person_ids: {person_ids_found[:3]}, old_business_id: {old_business_id})")
|
||
|
||
if not person_id:
|
||
# برای فاکتورهای فروش/خرید person_id الزامی است
|
||
if new_invoice_type in ["invoice_sales", "invoice_purchase",
|
||
"invoice_sales_return", "invoice_purchase_return"]:
|
||
# لاگ برای دیباگ
|
||
if self.stats["invoices_skipped"] < 5:
|
||
old_person_ids_str = ", ".join([str(pid) for pid in person_ids_found[:3]])
|
||
print(f" ⚠️ فاکتور {old_code} رد شد: person_id پیدا نشد (doc_id: {old_invoice.get('id')}, old_person_ids: {old_person_ids_str})")
|
||
self.stats["invoices_skipped"] += 1
|
||
return None
|
||
|
||
# استخراج سطرهای کالا
|
||
invoice_lines = []
|
||
skipped_commodities = []
|
||
for row in old_rows:
|
||
commodity_id = row.get('commodity_id')
|
||
if commodity_id:
|
||
product_id = product_mapping.get((old_business_id, commodity_id))
|
||
if not product_id:
|
||
skipped_commodities.append(commodity_id)
|
||
continue
|
||
|
||
# محاسبه مقادیر
|
||
debit = convert_amount(row.get('debit'))
|
||
credit = convert_amount(row.get('credit'))
|
||
quantity = convert_amount(row.get('quantity') or 1)
|
||
|
||
if quantity == 0:
|
||
quantity = Decimal(1)
|
||
|
||
# unit_price
|
||
unit_price = (debit + credit) / quantity if quantity > 0 else Decimal(0)
|
||
|
||
# تخفیف و مالیات
|
||
discount_amount = convert_amount(row.get('discount') or 0)
|
||
tax_amount = convert_amount(row.get('tax') or 0)
|
||
|
||
# محاسبه tax_percent
|
||
taxable_amount = (quantity * unit_price) - discount_amount
|
||
tax_rate = Decimal(0)
|
||
if taxable_amount > 0 and tax_amount > 0:
|
||
tax_rate = (tax_amount / taxable_amount) * 100
|
||
|
||
line = {
|
||
"product_id": product_id,
|
||
"quantity": float(quantity),
|
||
"extra_info": {
|
||
"unit_price": float(unit_price),
|
||
"line_discount": float(discount_amount),
|
||
"tax_amount": float(tax_amount),
|
||
"tax_rate": float(tax_rate)
|
||
},
|
||
"description": row.get('description')
|
||
}
|
||
|
||
invoice_lines.append(line)
|
||
|
||
if not invoice_lines:
|
||
# لاگ برای دیباگ
|
||
if self.stats["invoices_skipped"] < 5:
|
||
skipped_str = ", ".join([str(cid) for cid in skipped_commodities[:3]])
|
||
print(f" ⚠️ فاکتور {old_code} رد شد: هیچ سطر کالایی پیدا نشد (doc_id: {old_invoice.get('id')}, skipped_commodities: {skipped_str})")
|
||
self.stats["invoices_skipped"] += 1
|
||
return None
|
||
|
||
if dry_run:
|
||
self.stats["invoices_migrated"] += 1
|
||
return None
|
||
|
||
# استفاده از سرویس create_invoice
|
||
try:
|
||
# Import سرویس
|
||
from app.services.invoice_service import create_invoice
|
||
|
||
# اطمینان از وجود سیاست درآمدزایی
|
||
self.ensure_policy_exists(new_business_id)
|
||
|
||
# ایجاد فاکتور
|
||
# person_id باید در extra_info قرار بگیرد یا مستقیماً در data
|
||
extra_info_data = {
|
||
"source": "migration",
|
||
"old_document_id": old_invoice.get('id'),
|
||
"old_code": old_code,
|
||
"person_id": int(person_id) # اضافه کردن person_id به extra_info
|
||
}
|
||
|
||
result = create_invoice(
|
||
db=self.new_db,
|
||
business_id=new_business_id,
|
||
user_id=new_user_id,
|
||
data={
|
||
"invoice_type": new_invoice_type,
|
||
"document_date": doc_date.isoformat(),
|
||
"currency_id": new_currency_id,
|
||
"person_id": int(person_id), # همچنین در data هم قرار می‌دهیم
|
||
"lines": invoice_lines,
|
||
"description": old_invoice.get('description'),
|
||
"is_proforma": old_invoice.get('is_preview') == 1,
|
||
"extra_info": extra_info_data
|
||
}
|
||
)
|
||
|
||
# commit در batch انجام می‌شود (در run_migration)
|
||
# فقط ID را برمی‌گردانیم
|
||
new_invoice_id = result.get("id")
|
||
if new_invoice_id:
|
||
# اضافه کردن به cache
|
||
self.migrated_old_doc_ids.add(old_document_id)
|
||
if old_code:
|
||
cache_key = (new_business_id, old_code)
|
||
self.migrated_codes_cache[cache_key] = True
|
||
|
||
self.stats["invoices_migrated"] += 1
|
||
|
||
return new_invoice_id
|
||
|
||
except Exception as e:
|
||
# rollback و ایجاد session جدید
|
||
try:
|
||
self.new_db.rollback()
|
||
except Exception:
|
||
pass
|
||
|
||
# ایجاد session جدید برای ادامه کار
|
||
try:
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import sessionmaker
|
||
dsn = f"mysql+pymysql://root:136431@localhost:3306/{self.new_db_name}"
|
||
engine = create_engine(dsn, echo=False, pool_pre_ping=True)
|
||
self.new_db = sessionmaker(bind=engine)()
|
||
except Exception:
|
||
pass
|
||
|
||
self.stats["errors"] += 1
|
||
|
||
# استخراج جزئیات خطا از ApiError
|
||
error_msg = str(e)
|
||
error_code = None
|
||
if hasattr(e, 'code'):
|
||
error_code = e.code
|
||
if hasattr(e, 'message'):
|
||
error_msg = e.message
|
||
|
||
# بررسی خطای duplicate
|
||
if "Duplicate entry" in error_msg or "1062" in error_msg:
|
||
error_msg = "کد تکراری در warehouse_documents"
|
||
|
||
error_detail = {
|
||
"old_invoice_id": old_invoice.get('id'),
|
||
"old_code": old_code,
|
||
"error": error_msg,
|
||
"error_code": error_code
|
||
}
|
||
|
||
# فقط خطاهای اولیه را در لاگ نمایش می‌دهیم
|
||
if len(self.stats["error_details"]) < 10:
|
||
short_error = error_msg[:200] if len(error_msg) > 200 else error_msg
|
||
print(f" ❌ خطا در فاکتور {old_code} (ID: {old_invoice.get('id')}): {short_error}")
|
||
|
||
self.stats["error_details"].append(error_detail)
|
||
return None
|
||
|
||
except Exception as e:
|
||
self.stats["errors"] += 1
|
||
|
||
import traceback
|
||
error_trace = traceback.format_exc()
|
||
|
||
# فقط خطاهای اولیه را در لاگ نمایش می‌دهیم
|
||
if len(self.stats["error_details"]) < 10:
|
||
print(f" ❌ خطای عمومی در فاکتور {old_invoice.get('id', 'unknown')}: {str(e)}")
|
||
if "PERSON_REQUIRED" in str(e) or "person_id" in str(e).lower():
|
||
print(f" -> مشکل در person_id")
|
||
|
||
self.stats["error_details"].append({
|
||
"old_invoice_id": old_invoice.get('id'),
|
||
"error": str(e),
|
||
"traceback": error_trace[:500] if len(error_trace) > 500 else error_trace
|
||
})
|
||
return None
|
||
|
||
def run_migration(self, invoice_types: List[str] = None, dry_run: bool = False,
|
||
batch_size: int = 100, start_id: Optional[int] = None,
|
||
limit: Optional[int] = None, business_ids: Optional[List[int]] = None):
|
||
"""اجرای انتقال"""
|
||
if invoice_types is None:
|
||
invoice_types = ["sell", "buy", "rfsell", "rfbuy"]
|
||
|
||
print("=" * 100)
|
||
print("🚀 شروع انتقال فاکتورها")
|
||
print("=" * 100)
|
||
|
||
if dry_run:
|
||
print("\n⚠️ حالت DRY RUN - هیچ تغییری اعمال نمی‌شود")
|
||
|
||
# ایجاد mapping ها
|
||
print("\n📊 ایجاد mapping ها...")
|
||
business_id_mapping = self.create_business_id_mapping()
|
||
if not business_id_mapping:
|
||
print("❌ هیچ mapping کسب و کاری یافت نشد!")
|
||
return
|
||
|
||
user_id_mapping = self.create_user_id_mapping()
|
||
fiscal_year_mapping = self.create_fiscal_year_mapping(business_id_mapping)
|
||
currency_mapping = self.create_currency_mapping()
|
||
person_mapping = self.create_person_id_mapping(business_id_mapping)
|
||
product_mapping = self.create_product_id_mapping(business_id_mapping)
|
||
|
||
# انتقال برای هر نوع
|
||
for invoice_type in invoice_types:
|
||
print(f"\n{'='*100}")
|
||
print(f"📦 انتقال فاکتورهای نوع: {invoice_type}")
|
||
print(f"{'='*100}")
|
||
|
||
offset = 0
|
||
batch_commit_size = 10 # Commit هر 10 فاکتور
|
||
processed_in_batch = 0
|
||
|
||
# Load migrated documents cache در ابتدای هر نوع
|
||
self._load_migrated_documents_cache(business_id_mapping)
|
||
|
||
while True:
|
||
invoices = self.get_old_invoices(
|
||
invoice_type=invoice_type,
|
||
start_id=start_id if offset == 0 else None,
|
||
limit=batch_size,
|
||
offset=offset if offset > 0 else None,
|
||
business_ids=list(business_id_mapping.keys()) if not business_ids else business_ids
|
||
)
|
||
|
||
if not invoices:
|
||
break
|
||
|
||
print(f"\n📄 پردازش batch {offset // batch_size + 1}: {len(invoices)} فاکتور")
|
||
|
||
# Bulk load invoice rows
|
||
doc_ids = [inv.get('id') for inv in invoices]
|
||
rows_dict = self.get_invoice_rows_bulk(doc_ids) if not dry_run else {}
|
||
|
||
# Bulk duplicate check
|
||
if not dry_run:
|
||
self._batch_duplicate_check(invoices, business_id_mapping)
|
||
|
||
for invoice in invoices:
|
||
self.stats["invoices_processed"] += 1
|
||
|
||
# استفاده از rows از cache
|
||
if not dry_run and invoice.get('id') in rows_dict:
|
||
invoice['_cached_rows'] = rows_dict[invoice.get('id')]
|
||
|
||
if self.stats["invoices_processed"] % 10 == 0:
|
||
print(f" پردازش شده: {self.stats['invoices_processed']}, منتقل شده: {self.stats['invoices_migrated']}, رد شده: {self.stats['invoices_skipped']}, خطا: {self.stats['errors']}")
|
||
|
||
result = self.migrate_invoice(
|
||
old_invoice=invoice,
|
||
business_id_mapping=business_id_mapping,
|
||
user_id_mapping=user_id_mapping,
|
||
fiscal_year_mapping=fiscal_year_mapping,
|
||
currency_mapping=currency_mapping,
|
||
person_mapping=person_mapping,
|
||
product_mapping=product_mapping,
|
||
dry_run=dry_run
|
||
)
|
||
|
||
if result and not dry_run:
|
||
processed_in_batch += 1
|
||
# Batch commit
|
||
if processed_in_batch >= batch_commit_size:
|
||
try:
|
||
self.new_db.commit()
|
||
processed_in_batch = 0
|
||
except Exception as e:
|
||
self.new_db.rollback()
|
||
print(f" ⚠️ خطا در commit batch: {str(e)}")
|
||
|
||
# Commit نهایی batch
|
||
if not dry_run and processed_in_batch > 0:
|
||
try:
|
||
self.new_db.commit()
|
||
processed_in_batch = 0
|
||
except Exception as e:
|
||
self.new_db.rollback()
|
||
print(f" ⚠️ خطا در commit نهایی batch: {str(e)}")
|
||
|
||
offset += batch_size
|
||
|
||
if limit and self.stats["invoices_processed"] >= limit:
|
||
break
|
||
|
||
if len(invoices) < batch_size:
|
||
break
|
||
|
||
# نمایش آمار نهایی
|
||
print("\n" + "=" * 100)
|
||
print("📊 آمار نهایی")
|
||
print("=" * 100)
|
||
print(f"✅ فاکتورهای پردازش شده: {self.stats['invoices_processed']:,}")
|
||
print(f"✅ فاکتورهای منتقل شده: {self.stats['invoices_migrated']:,}")
|
||
print(f"⚠️ فاکتورهای رد شده: {self.stats['invoices_skipped']:,}")
|
||
print(f"❌ خطاها: {self.stats['errors']:,}")
|
||
|
||
if self.stats['error_details']:
|
||
print(f"\n⚠️ جزئیات خطاها (نمونه 10 مورد اول):")
|
||
for error in self.stats['error_details'][:10]:
|
||
print(f" - Invoice ID: {error.get('old_invoice_id')}, Code: {error.get('old_code')}")
|
||
print(f" Error: {error.get('error')}")
|
||
|
||
def close(self):
|
||
"""بستن اتصالات"""
|
||
self.old_db.close()
|
||
self.new_db.close()
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="انتقال فاکتورهای فروش، خرید و برگشت")
|
||
parser.add_argument("--dry-run", action="store_true", help="اجرای تست بدون تغییر در دیتابیس")
|
||
parser.add_argument("--batch-size", type=int, default=100, help="تعداد فاکتورها در هر batch")
|
||
parser.add_argument("--start-id", type=int, help="شروع از شناسه خاص")
|
||
parser.add_argument("--limit", type=int, help="محدود کردن تعداد فاکتورها")
|
||
parser.add_argument("--invoice-types", nargs="+",
|
||
choices=["sell", "buy", "rfsell", "rfbuy"],
|
||
default=["sell", "buy", "rfsell", "rfbuy"],
|
||
help="انواع فاکتور برای انتقال")
|
||
parser.add_argument("--old-db", default="hesabixOld", help="نام دیتابیس قدیمی")
|
||
parser.add_argument("--new-db", default="hesabixpy", help="نام دیتابیس جدید")
|
||
parser.add_argument("--db-user", default="root", help="نام کاربری دیتابیس")
|
||
parser.add_argument("--db-password", default="136431", help="رمز عبور دیتابیس")
|
||
parser.add_argument("--db-host", default="localhost", help="آدرس دیتابیس")
|
||
parser.add_argument("--db-port", type=int, default=3306, help="پورت دیتابیس")
|
||
|
||
args = parser.parse_args()
|
||
|
||
migration = InvoiceMigration(
|
||
old_db_name=args.old_db,
|
||
new_db_name=args.new_db,
|
||
db_user=args.db_user,
|
||
db_password=args.db_password,
|
||
db_host=args.db_host,
|
||
db_port=args.db_port
|
||
)
|
||
|
||
try:
|
||
migration.run_migration(
|
||
invoice_types=args.invoice_types,
|
||
dry_run=args.dry_run,
|
||
batch_size=args.batch_size,
|
||
start_id=args.start_id,
|
||
limit=args.limit
|
||
)
|
||
finally:
|
||
migration.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|