dddennnisss/trueconf_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: lk_diagnostic.py - fix strip syntax error

Browse Source
This commit is contained in:
Денис Кривоченко
2026-08-11 11:26:53 +07:00
parent 9e34b0597b
commit d2e7dd6234
1 changed files with 71 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lk/test/lk_diagnostic.py
+71 -68
View File
@@ -5,143 +5,146 @@ import os
import sys import sys
import pymssql import pymssql
BASE_DIR = '/opt/trueconf_bot' BASE_DIR = "/opt/trueconf_bot"
os.chdir(BASE_DIR) os.chdir(BASE_DIR)
def load_env(filepath): def load_env(filepath):
with open(filepath, 'r', encoding='utf-8') as f: with open(filepath, "r", encoding="utf-8") as f:
for line in f: for line in f:
line = line.strip() line = line.strip()
if line and not line.startswith('#') and '=' in line: if line and not line.startswith("#") and "=" in line:
key, value = line.split('=', 1) key, value = line.split("=", 1)
os.environ[key.strip()] = value.strip(' "'\''\r\n') value = value.strip(' "')
value = value.strip("'")
value = value.strip()
os.environ[key.strip()] = value
load_env(os.path.join(BASE_DIR, 'config', '.env')) load_env(os.path.join(BASE_DIR, "config", ".env"))
for pw_path in ['/opt/passwork', os.path.join(BASE_DIR, 'passwork')]: for pw_path in ["/opt/passwork", os.path.join(BASE_DIR, "passwork")]:
if os.path.isdir(pw_path) and pw_path not in sys.path: if os.path.isdir(pw_path) and pw_path not in sys.path:
sys.path.insert(0, pw_path) sys.path.insert(0, pw_path)
from passwork import get_passwork_secrets from passwork import get_passwork_secrets
card_name = os.getenv('PW_ID_SQL', '').strip() card_name = os.getenv("PW_ID_SQL", "").strip()
passwork_pool = get_passwork_secrets(required_cards=[card_name]) passwork_pool = get_passwork_secrets(required_cards=[card_name])
card_data = passwork_pool[card_name] card_data = passwork_pool[card_name]
DB_USER = card_data['login'] DB_USER = card_data["login"]
DB_PASSWORD = card_data['password'] DB_PASSWORD = card_data["password"]
SQL_SERVER = os.getenv('SQL_SERVER', 'SRVKEM-MOBILEIN.sibcem.ru') SQL_SERVER = os.getenv("SQL_SERVER", "SRVKEM-MOBILEIN.sibcem.ru")
SQL_DB_NAME = os.getenv('SQL_DB_NAME', 'BossCopy') SQL_DB_NAME = os.getenv("SQL_DB_NAME", "BossCopy")
conn = pymssql.connect( conn = pymssql.connect(
server=SQL_SERVER, server=SQL_SERVER,
user=DB_USER, user=DB_USER,
password=DB_PASSWORD, password=DB_PASSWORD,
database=SQL_DB_NAME, database=SQL_DB_NAME,
charset='cp1251', charset="cp1251",
login_timeout=10 login_timeout=10
) )
cursor = conn.cursor() cursor = conn.cursor()
email = 'la.guseynova@sibcem.ru' email = "la.guseynova@sibcem.ru"
print('=' * 60) print("=" * 60)
print(f'Diagnostika sotrudnika: {email}') print(f"Diagnostika sotrudnika: {email}")
print('=' * 60) print("=" * 60)
# 1. CARD_ID # 1. CARD_ID
print('\n--- 1. CARD_ID (iz PR_CARD + PR_EMP) ---') print("\n--- 1. CARD_ID (iz PR_CARD + PR_EMP) ---")
cursor.execute( cursor.execute(
'SELECT c.ID, e.EMAIL_ID FROM UOV_SELFSERVICE_PR_CARD c ' "SELECT c.ID, e.EMAIL_ID FROM UOV_SELFSERVICE_PR_CARD c "
'JOIN UOV_SELFSERVICE_PR_EMP e ON e.CARD_ID = c.ID ' "JOIN UOV_SELFSERVICE_PR_EMP e ON e.CARD_ID = c.ID "
'WHERE LOWER(e.EMAIL_ID) LIKE LOWER(%s)', "WHERE LOWER(e.EMAIL_ID) LIKE LOWER(%s)",
(f'%{email.split(@)[0]}@',) (f"%{email.split('@')[0]}@",)
) )
card_rows = cursor.fetchall() card_rows = cursor.fetchall()
if not card_rows: if not card_rows:
cursor.execute( cursor.execute(
'SELECT c.ID, e.EMAIL_ID FROM UOV_SELFSERVICE_PR_CARD c ' "SELECT c.ID, e.EMAIL_ID FROM UOV_SELFSERVICE_PR_CARD c "
'JOIN UOV_SELFSERVICE_PR_EMP e ON e.CARD_ID = c.ID ' "JOIN UOV_SELFSERVICE_PR_EMP e ON e.CARD_ID = c.ID "
'WHERE LOWER(e.EMAIL_ID) LIKE LOWER(%s)', "WHERE LOWER(e.EMAIL_ID) LIKE LOWER(%s)",
(f'%{email.split(@)[0]}%',) (f"%{email.split('@')[0]}%",)
) )
card_rows = cursor.fetchall() card_rows = cursor.fetchall()
if card_rows: if card_rows:
for r in card_rows: for r in card_rows:
print(f' CARD_ID: {r[0]}, EMAIL: {r[1]}') print(f" CARD_ID: {r[0]}, EMAIL: {r[1]}")
card_id = card_rows[0][0] card_id = card_rows[0][0]
else: else:
print(f' NE NAIDEN po email {email}') print(f" NE NAIDEN po email {email}")
conn.close() conn.close()
sys.exit(0) sys.exit(0)
print(f'\n Ispol' + 'zuem CARD_ID: {card_id}') print(f"\n Ispol'zuem CARD_ID: {card_id}")
# 2. PR_EMP # 2. PR_EMP
print('\n--- 2. PR_EMP (profil sotrudnika) ---') print("\n--- 2. PR_EMP (profil sotrudnika) ---")
cursor.execute( cursor.execute(
'SELECT ID, FIRM_ID, EMAIL_ID FROM UOV_SELFSERVICE_PR_EMP WHERE CARD_ID = %s', "SELECT ID, FIRM_ID, EMAIL_ID FROM UOV_SELFSERVICE_PR_EMP WHERE CARD_ID = %s",
(card_id,) (card_id,)
) )
emp_rows = cursor.fetchall() emp_rows = cursor.fetchall()
if emp_rows: if emp_rows:
for r in emp_rows: for r in emp_rows:
print(f' EMP_ID: {r[0]}, FIRM_ID: {r[1]}, EMAIL: {r[2]}') print(f" EMP_ID: {r[0]}, FIRM_ID: {r[1]}, EMAIL: {r[2]}")
else: else:
print(' PUSTO - net zapisey v PR_EMP') print(" PUSTO - net zapisey v PR_EMP")
# 3. PR_TRANS # 3. PR_TRANS
print('\n--- 3. PR_TRANS (kadrovye naznacheniya) ---') print("\n--- 3. PR_TRANS (kadrovye naznacheniya) ---")
if emp_rows: if emp_rows:
emp_ids = [r[0] for r in emp_rows] emp_ids = [r[0] for r in emp_rows]
placeholders = ','.join(['%s'] * len(emp_ids)) placeholders = ",".join(["%s"] * len(emp_ids))
cursor.execute( cursor.execute(
f'SELECT EMP_ID, D_FROM, D_TO, TP_ABBR, SYSPAY_ID ' f"SELECT EMP_ID, D_FROM, D_TO, TP_ABBR, SYSPAY_ID "
f'FROM UOV_SELFSERVICE_PR_TRANS ' f"FROM UOV_SELFSERVICE_PR_TRANS "
f'WHERE EMP_ID IN ({placeholders}) ' f"WHERE EMP_ID IN ({placeholders}) "
f'ORDER BY D_FROM DESC', f"ORDER BY D_FROM DESC",
emp_ids emp_ids
) )
trans_rows = cursor.fetchall() trans_rows = cursor.fetchall()
if trans_rows: if trans_rows:
print(f' Naideno zapisей: {len(trans_rows)}') print(f" Naideno zapisей: {len(trans_rows)}")
for r in trans_rows[:10]: for r in trans_rows[:10]:
print(f' EMP_ID: {r[0]}, D_FROM: {r[1]}, D_TO: {r[2]}, TP: {r[3]}, SYSPAY: {r[4]}') print(f" EMP_ID: {r[0]}, D_FROM: {r[1]}, D_TO: {r[2]}, TP: {r[3]}, SYSPAY: {r[4]}")
else: else:
print(' PUSTO - net zapisей v PR_TRANS') print(" PUSTO - net zapisey v PR_TRANS")
else: else:
print(' Skip (net PR_EMP)') print(" Skip (net PR_EMP)")
# 4. SL_YTAX # 4. SL_YTAX
print('\n--- 4. SL_YTAX (periody s dannymi) ---') print("\n--- 4. SL_YTAX (periody s dannymi) ---")
cursor.execute( cursor.execute(
'SELECT DISTINCT TYEAR, TMONTH, CMONTH FROM UOV_SELFSERVICE_SL_YTAX ' "SELECT DISTINCT TYEAR, TMONTH, CMONTH FROM UOV_SELFSERVICE_SL_YTAX "
'WHERE CARD_ID = %s ORDER BY CMONTH DESC', "WHERE CARD_ID = %s ORDER BY CMONTH DESC",
(card_id,) (card_id,)
) )
tax_rows = cursor.fetchall() tax_rows = cursor.fetchall()
if tax_rows: if tax_rows:
print(f' Naideno periodov: {len(tax_rows)}') print(f" Naideno periodov: {len(tax_rows)}")
for r in tax_rows[:10]: for r in tax_rows[:10]:
print(f' {r[0]}-{r[1]:02d} (CMONTH={r[2]})') print(f" {r[0]}-{r[1]:02d} (CMONTH={r[2]})")
else: else:
print(' PUSTO - net periodov v SL_YTAX') print(" PUSTO - net periodov v SL_YTAX")
# 5. Proverka q_firm # 5. Proverka q_firm
print('\n--- 5. Proverka q_firm dlya poslednego perioda ---') print("\n--- 5. Proverka q_firm dlya poslednego perioda ---")
if tax_rows: if tax_rows:
latest_cmonth = tax_rows[0][2] latest_cmonth = tax_rows[0][2]
tyear = (latest_cmonth - 1) // 12 tyear = (latest_cmonth - 1) // 12
tmonth = (latest_cmonth - 1) % 12 + 1 tmonth = (latest_cmonth - 1) % 12 + 1
import calendar import calendar
_, last_day = calendar.monthrange(tyear, tmonth) _, last_day = calendar.monthrange(tyear, tmonth)
date_start = f'{tyear}-{tmonth:02d}-01' date_start = f"{tyear}-{tmonth:02d}-01"
date_end = f'{tyear}-{tmonth:02d}-{last_day:02d}' date_end = f"{tyear}-{tmonth:02d}-{last_day:02d}"
date_params = (date_start, date_start, date_end, date_end, date_end) date_params = (date_start, date_start, date_end, date_end, date_end)
print(f' Period: {tyear}-{tmonth:02d}') print(f" Period: {tyear}-{tmonth:02d}")
q_firm = ''' q_firm = """
select e.FIRM_ID, f.SNAME from (select e.id, e.FIRM_ID from UOV_SELFSERVICE_PR_EMP e select e.FIRM_ID, f.SNAME from (select e.id, e.FIRM_ID from UOV_SELFSERVICE_PR_EMP e
join UOV_SELFSERVICE_PR_CARD c on c.ID=e.CARD_ID where c.ID = %s join UOV_SELFSERVICE_PR_CARD c on c.ID=e.CARD_ID where c.ID = %s
group by e.ID, e.FIRM_ID ) as e group by e.ID, e.FIRM_ID ) as e
@@ -149,39 +152,39 @@ if tax_rows:
join UOV_SELFSERVICE_HR_FIRM f on f.ID = e.FIRM_ID join UOV_SELFSERVICE_HR_FIRM f on f.ID = e.FIRM_ID
where ((t.D_FROM <= %s and t.D_TO >= %s) or (t.D_FROM <= %s and t.D_TO >= %s) or (t.D_FROM <= %s and t.D_TO = '1999-12-31')) where ((t.D_FROM <= %s and t.D_TO >= %s) or (t.D_FROM <= %s and t.D_TO >= %s) or (t.D_FROM <= %s and t.D_TO = '1999-12-31'))
group by e.FIRM_ID, f.SNAME group by e.FIRM_ID, f.SNAME
''' """
cursor.execute(q_firm, (card_id, *date_params)) cursor.execute(q_firm, (card_id, *date_params))
firm_row = cursor.fetchone() firm_row = cursor.fetchone()
if firm_row: if firm_row:
print(f' q_firm NAISHEL: FIRM_ID={firm_row[0]}, SNAME={firm_row[1]}') print(f" q_firm NAISHEL: FIRM_ID={firm_row[0]}, SNAME={firm_row[1]}")
else: else:
print(f' q_firm NE NAISHEL (NULL) - eto vyzovet oshibku!') print(f" q_firm NE NAISHEL (NULL) - eto vyzovet oshibku!")
q_firm_fb = ''' q_firm_fb = """
select e.FIRM_ID from UOV_SELFSERVICE_PR_EMP e select e.FIRM_ID from UOV_SELFSERVICE_PR_EMP e
join UOV_SELFSERVICE_PR_CARD c on c.ID=e.CARD_ID join UOV_SELFSERVICE_PR_CARD c on c.ID=e.CARD_ID
where c.ID = %s where c.ID = %s
group by e.ID, e.FIRM_ID group by e.ID, e.FIRM_ID
''' """
cursor.execute(q_firm_fb, (card_id,)) cursor.execute(q_firm_fb, (card_id,))
fb_row = cursor.fetchone() fb_row = cursor.fetchone()
if fb_row: if fb_row:
print(f' Fallbek NAISHEL: FIRM_ID={fb_row[0]}') print(f" Fallbek NAISHEL: FIRM_ID={fb_row[0]}")
else: else:
print(f' Fallbek NE NAISHEL - kriticheskaya oshibka!') print(f" Fallbek NE NAISHEL - kriticheskaya oshibka!")
# 6. Svodka # 6. Svodka
print('\n--- 6. Svodka ---') print("\n--- 6. Svodka ---")
cursor.execute( cursor.execute(
'SELECT COUNT(DISTINCT CMONTH), MIN(CMONTH), MAX(CMONTH) FROM UOV_SELFSERVICE_SL_YTAX WHERE CARD_ID = %s', "SELECT COUNT(DISTINCT CMONTH), MIN(CMONTH), MAX(CMONTH) FROM UOV_SELFSERVICE_SL_YTAX WHERE CARD_ID = %s",
(card_id,) (card_id,)
) )
summary = cursor.fetchone() summary = cursor.fetchone()
if summary: if summary:
print(f' Periodov v SL_YTAX: {summary[0]}') print(f" Periodov v SL_YTAX: {summary[0]}")
print(f' Diapazon: {summary[1]} - {summary[2]}') print(f" Diapazon: {summary[1]} - {summary[2]}")
print('\n' + '=' * 60) print("\n" + "=" * 60)
print('Dagnostika zavershena') print("Dagnostika zavershena")
print('=' * 60) print("=" * 60)
conn.close() conn.close()