import sys, os, json, time, re, requests, traceback, pymysql, datetime, string, random
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
if BASE_DIR not in sys.path: sys.path.insert(0, BASE_DIR)
LOG_FILE = os.path.join(BASE_DIR, 'history.log')
SESSION_FILE = os.path.join(BASE_DIR, 'bot_sessions.json')
BOT_TOKEN = "8690070480:AAHfl0bPp7XIEBXCxz07fIsHU3t0hCn29eY"
def load_tz_config():
offset_str = "+04:30"
h, m = 4, 30
dt_file = os.path.join(BASE_DIR, 'date-time.json')
if os.path.exists(dt_file):
try:
with open(dt_file, 'r') as f:
dt_cfg = json.load(f)
offset_str = dt_cfg.get('db_offset', '+04:30')
sign = -1 if offset_str.startswith('-') else 1
parts = offset_str.replace('+', '').replace('-', '').split(':')
h = sign * int(parts[0])
m = sign * int(parts[1]) if len(parts) > 1 else 0
except: pass
return offset_str, datetime.timezone(datetime.timedelta(hours=h, minutes=m))
DB_TZ_OFFSET, LOCAL_TZ = load_tz_config()
DB_CFG = {'host': 'localhost', 'user': 'evisuyfj_GlobalDeviceCheckerBot', 'password': 'GlobalDeviceCheckerBot', 'database': 'evisuyfj_GlobalDeviceCheckerBot', 'cursorclass': pymysql.cursors.DictCursor, 'autocommit': True, 'charset': 'utf8mb4'}
def write_history(stage, message):
try:
now = datetime.datetime.now(LOCAL_TZ).strftime('%Y-%m-%d %H:%M:%S')
with open(LOG_FILE, 'a', encoding='utf-8') as f:
f.write(f"[{now}] [SYSTEM_WORKER] [{stage}] {message}\n")
except: pass
def send_tg_notification(chat_id, text, kb=None):
try:
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
d = {'chat_id': chat_id, 'text': text, 'parse_mode': 'HTML'}
if kb: d['reply_markup'] = json.dumps(kb)
requests.post(url, json=d, verify=False, timeout=5)
except: pass
def request_manual_captcha(user_id, task_id, img_bytes, ck, token='', loc=''):
s = {}
if os.path.exists(SESSION_FILE):
try: s = json.load(open(SESSION_FILE))
except: pass
if str(user_id) not in s: s[str(user_id)] = {}
s[str(user_id)].update({
'st': 'AWAITING_MANUAL_CAP',
'task_id': task_id,
'ck': ck,
'token': token,
'loc': loc
})
with open(SESSION_FILE, 'w') as f: json.dump(s, f)
c = get_conn()
if c:
c.cursor().execute("UPDATE imei_logs SET status='AWAITING_CAPTCHA' WHERE id=%s", (task_id,))
c.commit()
c.close()
try:
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendPhoto"
files = {'photo': ('captcha.jpg', img_bytes)}
data = {'chat_id': user_id, 'caption': "š¤ Human Verification Required\n\nFor security reasons, please type the characters from the image above to process your order.", 'parse_mode': 'HTML'}
requests.post(url, data=data, files=files, verify=False, timeout=15)
except: pass
def clean_for_api(html_text):
text = re.sub(r'<[^>]+>', '', html_text)
text = text.replace('ā
Verification Successful', '--- VERIFICATION SUCCESSFUL ---')
emojis = ['ā
', 'ā', 'š', 'š', 'š
±ļø', 'š±', 'š²', 'š', 'āļø', 'š', 'š', 'š¢', 'š
', 'š ', 'āļø', 'š«', 'š', 'ā³', 'š', 'š°', 'š·', 'š', 'ā¶ļø']
for emj in emojis: text = text.replace(emj, '')
text = text.replace('ā', '-').replace('ā', '-')
text = re.sub(r'/Reports', '', text, flags=re.IGNORECASE)
text = re.sub(r'To check another device.*', '', text, flags=re.DOTALL | re.IGNORECASE)
return re.sub(r'\n{3,}', '\n\n', re.sub(r'\n\s+', '\n', text)).strip()
def format_ifree_obj(obj):
if not isinstance(obj, dict): return str(obj)
out = ""
for k, v in obj.items():
if k.lower() in ['imei', 'imei2', 'serial']: continue
if v == "" or v is None: continue
clean_k = str(k).replace('_', ' ').title()
out += f" ā {clean_k}: {v}\n"
return out
def get_ci(data_dict, key_lower, default_val='N/A'):
for k, v in data_dict.items():
if str(k).lower() == key_lower.lower():
return str(v).strip() if v is not None and str(v).strip() != "" else default_val
return default_val
try: from checkers import generic, xiaomi, apple, transsion, solver
except: pass
def get_conn():
try:
conn = pymysql.connect(**DB_CFG)
conn.cursor().execute(f"SET time_zone = '{DB_TZ_OFFSET}'")
return conn
except: return None
def gen_track_id(prefix):
chars = string.ascii_lowercase + string.digits
return f"{prefix}-" + ''.join(random.choices(chars, k=6))
def run_worker():
while True:
c = get_conn()
if not c: break
try:
c.begin()
cur = c.cursor()
cur.execute("""
SELECT l.*, s.api_code, s.service_name, s.use_auto_captcha
FROM imei_logs l
LEFT JOIN services s ON l.service_id = s.id
WHERE l.status IN ('PENDING', 'MANUAL_CAP_READY') AND l.source IN ('API', 'BOT')
ORDER BY l.created_at ASC LIMIT 1 FOR UPDATE
""")
task = cur.fetchone()
if not task: c.rollback(); c.close(); break
task_id, track_id, imei = task['id'], task['track_id'], task['imei']
expected_brand, user_id, cost = task['brand'], task['user_id'], float(task['cost'])
source, api_code, srv_name = task['source'], task['api_code'], task['service_name']
use_auto_cap = int(task['use_auto_captcha']) if task.get('use_auto_captcha') is not None else 1
is_manual_ready = (task['status'] == 'MANUAL_CAP_READY')
cur.execute("UPDATE imei_logs SET status = 'PROCESSING' WHERE id = %s", (task_id,))
c.commit()
except:
if c: c.rollback()
if c: c.close()
break
finally:
if c and c.open: c.close()
is_success, tg_base, raw_data_json, detected_model = False, "", None, "Unknown"
try:
# 1. ZORG CUSTOM XIAOMI ROUTER
if api_code == 'XIAOMI-FMI':
inf = generic.call_ifree(imei) or {}
fmi_status = xiaomi.call_mi_fmi(imei)
if fmi_status is not None:
is_success, raw_data_json = True, {"fmi": fmi_status}
detected_model = inf.get('model') or inf.get('product') or "Unknown"
manu = inf.get('manufacturer') or 'Xiaomi Communications Co Ltd'
tg_base = f"š Brand Information\n ā š Manufacturer: {manu}\n ā š
±ļø Brand: Xiaomi\n\nš FMI Status\n ā š² Model: {detected_model}\n ā š Find My Device: {'ON' if fmi_status else 'OFF'}"
else: tg_base = "Failed to fetch FMI status. Device may be invalid."
elif api_code == 'XIAOMI-BASIC':
inf = generic.call_ifree(imei)
if inf:
is_success, raw_data_json = True, inf
detected_model = inf.get('model') or "Unknown"
manu = inf.get('manufacturer') or 'Xiaomi Communications Co Ltd'
mcode = inf.get('modelName') or ''
tg_base = f"š Brand Information\n ā š Manufacturer: {manu}\n ā š
±ļø Brand: Xiaomi\n\nš Device Info\n ā š² Model: {detected_model}\n"
if mcode.strip() and mcode.upper() != "N/A":
tg_base += f" ā š Model Code: {mcode}"
else:
tg_base = tg_base.replace(" ā š² Model:", " ā š² Model:", 1)
else: tg_base = "Failed to fetch Basic Info."
elif api_code == 'XIAOMI-OFFICIAL' or (expected_brand.upper() == 'XIAOMI' and api_code in [None, '', 'legacy']):
if is_manual_ready:
s_data = json.load(open(SESSION_FILE)).get(str(user_id), {})
ck = s_data.get('ck', {})
captcha_text = s_data.get('cap_val', '')
img = b"dummy"
else:
img, ck = xiaomi.get_cap()
captcha_text = None
if img and use_auto_cap == 1:
captcha_text = solver.solve_image(img)
if img:
if not captcha_text and not is_manual_ready:
request_manual_captcha(user_id, task_id, img, ck)
continue
if captcha_text:
res = xiaomi.call_mi(imei, captcha_text, ck)
if res and res.get('code') == 1:
is_success, raw_data_json = True, res
xd = res.get('data', {})
fmi_status = xiaomi.call_mi_fmi(imei)
locked = fmi_status if fmi_status is not None else xd.get('locked', False)
region = xd.get('country_text') or 'Unknown'
mfg_date_str = ""
if xd.get('add_time'):
try: mfg_date_str = datetime.datetime.fromtimestamp(int(xd.get('add_time'))).strftime('%Y-%m-%d')
except: pass
inf = generic.call_ifree(imei) or {}
prod = xd.get('goods_name') or inf.get('product') or 'Unknown'
detected_model = xd.get('model') or inf.get('model') or 'Unknown'
manu = inf.get('manufacturer') or 'Xiaomi Communications Co Ltd'
mcode = inf.get('modelName') or ''
tg_base = f"š Brand Information\n ā š Manufacturer: {manu}\n ā š
±ļø Brand: Xiaomi\n\nš Device Info\n"
if prod != "Unknown": tg_base += f" ā š± Product Name: {prod}\n"
tg_base += f" ā š² Model: {detected_model}\n"
if mcode.strip() and mcode.upper() != "N/A":
tg_base += f" ā š Model Code: {mcode}\n"
if mfg_date_str:
tg_base += f" ā āļø Production Date: {mfg_date_str}\n"
else:
tg_base = tg_base.replace(" ā š", " ā š", 1) if "š" in tg_base else tg_base.replace(" ā š²", " ā š²", 1)
tg_base += f"\nš More Info\n ā š Region: {region}\n ā š Find My Device: {'ON' if locked else 'OFF'}"
else: tg_base = "Xiaomi Official Check Failed or Invalid IMEI."
else: tg_base = "Failed to process security image. Please try again."
else: tg_base = "Could not reach Xiaomi Official Servers."
elif api_code == 'XIAOMI-PRO':
c_conf = get_conn()
cur_conf = c_conf.cursor()
cur_conf.execute("SELECT setting_value FROM settings WHERE setting_key='ifree_api_key'")
row = cur_conf.fetchone()
c_conf.close()
api_key = row['setting_value'] if row else "2JE-6C0-BLR-7WI-QVB-ZHZ-3GV-6ZI"
payload = {"key": api_key, "service": "196", "imei": imei}
req = requests.post("https://api.ifreeicloud.co.uk", data=payload, timeout=60).json()
if is_manual_ready:
s_data = json.load(open(SESSION_FILE)).get(str(user_id), {})
ck = s_data.get('ck', {})
captcha_text = s_data.get('cap_val', '')
img = b"dummy"
else:
img, ck = xiaomi.get_cap()
captcha_text = None
if img and use_auto_cap == 1:
captcha_text = solver.solve_image(img)
locked_status = 'OFF'
mfg_date_official = None
if img:
if not captcha_text and not is_manual_ready:
request_manual_captcha(user_id, task_id, img, ck)
continue
if captcha_text:
res = xiaomi.call_mi(imei, captcha_text, ck)
if res and res.get('code') == 1:
xd = res.get('data', {})
if xd.get('add_time'):
try: mfg_date_official = datetime.datetime.fromtimestamp(int(xd.get('add_time'))).strftime('%Y-%m-%d')
except: pass
fmi_status = xiaomi.call_mi_fmi(imei)
if fmi_status is not None: locked_status = 'ON' if fmi_status else 'OFF'
if req.get('success'):
is_success = True
raw_data_json = req.get('object', {})
p_name = get_ci(raw_data_json, 'modelname', 'Unknown')
m_code = get_ci(raw_data_json, 'modelcode')
sku = get_ci(raw_data_json, 'skunumber')
unlock_num = get_ci(raw_data_json, 'unlocknumber')
p_country = get_ci(raw_data_json, 'purchasecountry')
a_country = get_ci(raw_data_json, 'activationcountry')
w_status = get_ci(raw_data_json, 'warrantystatus')
w_desc = get_ci(raw_data_json, 'warrantydescription')
w_start = get_ci(raw_data_json, 'warrantystartdate')
w_end = get_ci(raw_data_json, 'warrantyenddate')
d_date = get_ci(raw_data_json, 'deliverydate')
a_date = get_ci(raw_data_json, 'activationdate')
prod_date = mfg_date_official if mfg_date_official else get_ci(raw_data_json, 'productiondate')
imei1 = get_ci(raw_data_json, 'imei', imei)
imei2 = get_ci(raw_data_json, 'imei2')
serial = get_ci(raw_data_json, 'serial')
if a_date != 'N/A' and len(a_date) > 10: a_date = a_date.split(' ')[0]
if w_start != 'N/A' and len(w_start) > 10: w_start = w_start.split(' ')[0]
if d_date != 'N/A' and len(d_date) > 10: d_date = d_date.split(' ')[0]
if prod_date != 'N/A' and len(prod_date) > 10 and ' ' in prod_date: prod_date = prod_date.split(' ')[0]
inf = generic.call_ifree(imei) or {}
detected_model = inf.get('model')
if not detected_model and p_name != 'Unknown':
detected_model = " ".join(p_name.split(' ')[:3])
elif not detected_model:
detected_model = 'Unknown'
manu = inf.get('manufacturer') or 'Xiaomi Communications Co Ltd'
final_mcode = m_code if m_code != 'N/A' else sku
tg_base = f"š Brand Information\n ā š Manufacturer: {manu}\n ā š
±ļø Brand: Xiaomi\n\n"
tg_base += f"š Device Info\n ā š± Product Name: {p_name}\n ā š² Model: {detected_model}\n"
if final_mcode and final_mcode != 'N/A': tg_base += f" ā š Model Code: {final_mcode}\n"
if sku and sku != 'N/A': tg_base += f" ā š¦ SKU Number: {sku}\n"
if unlock_num and unlock_num != 'N/A': tg_base += f" ā š Unlock Number: {unlock_num}\n"
tg_base += f" ā āļø Production Date: {prod_date}\n\n"
tg_base += f"š Purchase & Activation\n ā š Purchase Country: {p_country}\n"
tg_base += f" ā š Activation Country: {a_country}\n"
tg_base += f" ā š Delivery Date: {d_date}\n"
tg_base += f" ā š
Activation Date: {a_date}\n\n"
tg_base += f"š Warranty Info\n ā š” Warranty Status: {w_status}\n ā š Warranty Desc: {w_desc}\n ā ā³ Warranty Start: {w_start}\n ā āļø Warranty End: {w_end}\n\n"
tg_base += f"š Security Status\n ā š Find My Device: {locked_status}\n\n"
tg_base += f"š Identify Info\n ā š¢ IMEI 1: {imei1}\n"
id_list = []
if imei2 != 'N/A': id_list.append(f"š¢ IMEI 2: {imei2}")
if serial != 'N/A': id_list.append(f"š¢ Serial Number: {serial}")
for i, item in enumerate(id_list):
prefix = " ā " if i == len(id_list) - 1 else " ā "
tg_base += f"{prefix}{item}\n"
if not id_list:
tg_base = tg_base.replace(" ā š¢ IMEI 1", " ā š¢ IMEI 1")
parsed_keys_lower = ['modelname', 'modelcode', 'skunumber', 'unlocknumber', 'purchasecountry', 'activationcountry', 'warrantystatus', 'warrantydescription', 'warrantystartdate', 'warrantyenddate', 'deliverydate', 'activationdate', 'productiondate', 'miactivationlock', 'imei', 'imei2', 'serial', 'searchterm']
unparsed_data = {}
for k, v in raw_data_json.items():
if str(k).lower() not in parsed_keys_lower and v and str(v).strip() != "" and str(v).strip().upper() != "N/A":
unparsed_data[k] = v
if unparsed_data:
tg_base += "\n\nš Other Info\n"
keys_list = list(unparsed_data.keys())
for i, k in enumerate(keys_list):
prefix = " ā " if i == len(keys_list) - 1 else " ā "
clean_k = str(k).replace('_', ' ').title()
tg_base += f"{prefix}{clean_k}: {unparsed_data[k]}\n"
else:
tg_base = f"API Error: {req.get('error', 'Unknown Error')}"
# 2. iFREE API ROUTER
elif api_code and api_code.startswith('IFREE-'):
ifree_id = api_code.split('-')[1]
c_conf = get_conn()
cur_conf = c_conf.cursor()
cur_conf.execute("SELECT setting_value FROM settings WHERE setting_key='ifree_api_key'")
row = cur_conf.fetchone()
c_conf.close()
api_key = row['setting_value'] if row else "2JE-6C0-BLR-7WI-QVB-ZHZ-3GV-6ZI"
if ifree_id == '0':
res = generic.call_ifree(imei)
if res:
is_success, raw_data_json = True, res
detected_model = res.get('model') or res.get('product') or "Unknown"
tg_base = f"š Basic Information\n{format_ifree_obj(res)}"
else: tg_base = "Basic Check Failed."
else:
payload = {"key": api_key, "service": ifree_id, "imei": imei}
req = requests.post("https://api.ifreeicloud.co.uk", data=payload, timeout=60).json()
if req.get('success'):
is_success = True
raw_data_json = req.get('object', {})
detected_model = raw_data_json.get('modelDescription') or raw_data_json.get('model') or expected_brand
tg_base = f"š {srv_name}\n{format_ifree_obj(raw_data_json)}"
else: tg_base = f"API Error: {req.get('error', 'Unknown Error')}"
# 3. LEGACY / FALLBACK
elif expected_brand.upper() == 'TRANSSION':
res = transsion.process(imei)
if 'Verification Successful' in res['text']:
is_success, raw_data_json = True, res['raw']
m = re.search(r'š²\s*Model:\s*(.*?)', res['text'])
detected_model = m.group(1) if m else "Transsion Device"
tg_base = re.sub(r'To check another device.*?@GlobalDeviceCheckerBot', '', res['text'], flags=re.DOTALL).strip()
else: tg_base = "Failed to fetch Transsion details."
elif expected_brand.upper() == 'APPLE':
if is_manual_ready:
s_data = json.load(open(SESSION_FILE)).get(str(user_id), {})
ck = s_data.get('ck', {})
token = s_data.get('token', '')
loc = s_data.get('loc', 'en_US')
captcha_text = s_data.get('cap_val', '')
img = b"dummy"
else:
cap_data = apple.get_apple_cap()
if cap_data and len(cap_data) == 4:
img, ck, token, loc = cap_data
else:
img, ck, token, loc = None, {}, '', ''
captcha_text = None
if img and use_auto_cap == 1:
captcha_text = solver.solve_image(img)
if img:
if not captcha_text and not is_manual_ready:
request_manual_captcha(user_id, task_id, img, ck, token, loc)
continue
if captcha_text:
res = apple.call_apple(imei, captcha_text, ck, token, loc)
if res and res.get('code') == 1:
is_success, raw_data_json = True, res.get('raw')
xd, ifree = res.get('data', {}), generic.call_ifree(imei)
prod = ifree.get('product') or ifree.get('model') or xd.get('goods_name') or 'Unknown Device'
brand_safe = generic.get_dynamic_brand(prod, "Apple")
detected_model = xd.get('goods_name') or ifree.get('model') or brand_safe
tg_base = f"š Brand Information\n ā š Manufacturer: Apple Inc.\n ā š
±ļø Brand: {brand_safe}\n\nš Device Info\n ā š± Product Name: {prod}\n ā š² Model: {detected_model}\n\nš Coverage Info\n ā š
Purchased: {xd.get('dop', 'N/A')}\n ā š
Expired at: {xd.get('expiry', 'N/A')}\n ā š
Date Now: {datetime.datetime.now().strftime('%B %d, %Y')}\n ā š Open Repair: {xd.get('open_repair', 'false')}\n ā š Coverage: {xd.get('coverage_status', 'Unknown')}"
else:
tg_base = "ā Apple Server Rejection\n\nā ļø The official Apple database rejected this Serial Number or failed to return valid coverage data.\n\nPossible Reasons:\n1. The Serial Number is incorrect or typoed.\n2. The device is a Replacement/Repacked model that requires a Premium Checker.\n3. The device is too old or too new to be in the main warranty database.\n\nš Suggestion: Please verify your Serial Number or try using one of the Premium Apple Services from the menu."
else: tg_base = "Failed to process security image. Please try again."
else: tg_base = "Could not get Apple CAPTCHA image."
else: tg_base = "Unsupported Brand or Service not defined."
except Exception as e:
tg_base = f"System Error: {str(e)}"
write_history("WORKER_CHECK_ERR", traceback.format_exc())
tg_base = tg_base.replace("ā
Verification Successful\n\n", "").replace("ā
Verification Successful\n", "")
tg_base = re.sub(r'To check another device.*?@GlobalDeviceCheckerBot', '', tg_base, flags=re.DOTALL).strip()
tg_base = re.sub(r'Auto-Solved via .*', '', tg_base, flags=re.IGNORECASE).strip()
name_clean = srv_name.replace(" - FREE", "") if srv_name else ('Paid (Full Info)' if cost > 0 else 'FREE (Basic Info)')
job_id_str = track_id if track_id else (f"PAID_{task_id}" if cost > 0 else f"FREE_{task_id}")
server_status = f"š Server Status\n ā š¢ Search Term: {imei}\n ā š Work ID: {job_id_str}\n ā ā¶ļø Action: {name_clean}\n"
if is_success:
tg_base = tg_base.strip()
final_tg = server_status + f" ā ā
Status: SUCCESS\n{tg_base}
\n/Reports\nTo check another device, just send the new code."
else:
tg_base = tg_base.strip()
refund_note = f"\n\nš° Refunded: ${cost:g}" if cost > 0 else ""
final_tg = server_status + f" ā ā Status: FAILED\n\n š¬ Reason: \n{tg_base[:1000]}{refund_note}\n\n/Reports\nTo check another device, just send the new code."
db_text_pure = clean_for_api(tg_base) if is_success else tg_base
c = get_conn()
if c:
try:
c.begin()
cur = c.cursor()
if is_success:
cur.execute("UPDATE imei_logs SET status='FULL', response_text=%s, raw_data=%s, device_model=%s WHERE id=%s", (db_text_pure, json.dumps(raw_data_json) if raw_data_json else None, detected_model, task_id))
else:
cur.execute("UPDATE imei_logs SET status='FAILED', response_text=%s WHERE id=%s", (db_text_pure, task_id))
if cost > 0:
cur.execute("UPDATE users SET credit = credit + %s WHERE user_id = %s", (cost, user_id))
trf_id = gen_track_id("TRF-AddPay")
cur.execute("INSERT INTO transactions (user_id, amount, type, note, created_at, track_id) VALUES (%s, %s, 'REFUND', %s, NOW(), %s)", (user_id, cost, f"Refund for {job_id_str}", trf_id))
c.commit()
if source == 'API': send_tg_notification(user_id, f"š API Order Completed!\n\nš Work ID: {job_id_str}\nā
Status: SUCCESS\n\n{tg_base}" if is_success else f"š API Order Failed!\n\nš Work ID: {job_id_str}\nā Reason: {tg_base}\nš° Refunded: ${cost:g}")
else: send_tg_notification(user_id, final_tg)
except: c.rollback()
finally: c.close()
if __name__ == "__main__":
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
run_worker()