import os
import json
import csv
import traceback
import re
import html
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, unquote, parse_qs
import IP2Location
from phonenumbers.phonenumberutil import NumberParseException
from phonenumbers import parse, geocoder
# --- Константы ---
MAX_URI_LENGTH = 4000
MAX_GET_TEXT_LENGTH = 2000
database = IP2Location.IP2Location(os.path.join("IP2LOCATION-LITE-DB3.BIN"))
russian_phone_db = None
# Частотные русские биграмы для оценки "осмысленности" текста
COMMON_BIGRAMS = {
'ст', 'но', 'на', 'ен', 'ть', 'ов', 'ей', 'пр', 'ли', 'ле', 'ро', 'по',
'ер', 'ни', 'ло', 'во', 'не', 'за', 'ко', 'то', 'ра', 'со', 'ре', 'от',
'ла', 'де', 'се', 'ми', 'ти', 'ди', 'ки', 'до', 'ма', 'ме', 'пе', 'ве',
'те', 'же', 'че', 'ще', 'шу', 'жа', 'ца', 'ча', 'ща', 'юб', 'юж', 'юн',
'яб', 'яв', 'яг', 'яд', 'яз', 'як', 'ям', 'ян', 'яп', 'яс', 'ящ', 'ия', 'ие'
}
def load_russian_phone_database():
global russian_phone_db
if russian_phone_db is None:
russian_phone_db = []
try:
csv_path = os.path.join(os.path.dirname(__file__), "DEF-9xx.csv")
if os.path.exists(csv_path):
with open(csv_path, "r", encoding='utf-8') as f:
reader = csv.reader(f, delimiter="\t")
for i, line in enumerate(reader):
if i != 0 and line:
parts = line[0].split(";")
if len(parts) >= 6:
zone = parts[0]
start_range = int(parts[1])
end_range = int(parts[2])
region = parts[5].strip()
russian_phone_db.append({
'zone': zone, 'start': start_range, 'end': end_range, 'region': region
})
print(f"Загружено {len(russian_phone_db)} диапазонов номеров телефонов из базы")
except Exception as e:
print(f"Ошибка загрузки базы телефонов: {e}")
return russian_phone_db
def get_region_by_ip(ip_address):
try:
rec = database.get_all(ip_address)
return {"region_name": rec.region}
except Exception:
return {"region_name": "не определен"}
def normalize_phone(phone):
return re.sub(r'[^\d+]', '', phone)
def get_region_by_phone_russia(zone, number):
phone_db = load_russian_phone_database()
if not phone_db:
return None
for record in phone_db:
if record['zone'] == zone and record['start'] <= number <= record['end']:
return record['region']
return None
def get_region_by_phone(phone):
normalized = normalize_phone(phone)
if normalized.startswith("+7") and len(normalized) >= 5 and normalized[2] == "9":
zone = normalized[2:5]
try:
number = int(normalized[5:])
region = get_region_by_phone_russia(zone, number)
if region: return {"region_name": region}
except ValueError: pass
elif normalized.startswith("8") and len(normalized) >= 4 and normalized[1] == "9":
zone = normalized[1:4]
try:
number = int(normalized[4:])
region = get_region_by_phone_russia(zone, number)
if region: return {"region_name": region}
except ValueError: pass
try:
parsed = parse(normalized)
region = geocoder.description_for_number(parsed, 'ru')
if region: return {"region_name": region}
except NumberParseException: pass
return {"region_name": "не определен"}
def detect_input_type(input_string):
parts = input_string.replace(' ', '').split('.')
if len(parts) == 4:
try:
for part in parts:
if not 0 <= int(part) <= 255: return "phone"
return "ip"
except ValueError: return "phone"
if ':' in input_string and '.' not in input_string: return "ip"
digits = sum(c.isdigit() for c in input_string)
return "phone" if digits >= 7 else "unknown"
# ---------- КОДИРОВКИ И ОЧИСТКА ----------
def count_cyrillic(text):
return len(re.findall(r'[А-Яа-яЁё]', text))
def text_quality_score(text: str) -> float:
"""Оценка читаемости: кириллица + биграмный анализ русского языка."""
if not text: return -1e9
clean = text.replace('\xa0', ' ').replace('\u00a0', ' ')
total = len(clean)
if total == 0: return 0
cyr = len(re.findall(r'[А-Яа-яЁё]', clean))
bad = sum(1 for ch in clean if ch == '\ufffd' or (ord(ch) < 32 and ch not in '\n\r\t'))
moji = len(re.findall(r'[\u00D0\u00D1][\x80-\xBF]', clean))
# Биграмный бонус: осмысленный текст содержит частые пары букв
bigrams = re.findall(r'[а-яё]{2}', clean.lower())
bigram_hit = sum(1 for bg in bigrams if bg in COMMON_BIGRAMS)
bigram_ratio = bigram_hit / max(total, 1)
score = (cyr / total) * 2.5 + bigram_ratio * 2.5 - (bad / total) * 5.0 - (moji / total) * 4.0
return score
def extract_html_charset(raw_text: str) -> str | None:
# Look for charset in meta tag as well as in content-type headers
# Match both and
meta_charset_match = re.search(r']+charset[\'"\s]*=[\'"\s]*([a-zA-Z0-9_-]+)', raw_text, re.IGNORECASE)
if meta_charset_match:
return meta_charset_match.group(1).lower()
content_type_match = re.search(r'charset=["\']?([a-zA-Z0-9_-]+)["\']?', raw_text, re.IGNORECASE)
if content_type_match:
return content_type_match.group(1).lower()
return None
def fix_mojibake_by_decoding(text: str) -> str:
"""
Универсальный декодер на основе абсолютного скоринга.
Автоматически выбирает CP1251, KOI8-R или UTF-8 без жестких порогов.
Возвращает оригинальный текст, если декодирование не улучшает результат.
"""
if not text or len(text) < 5: return text
if text.count('?') / len(text) > 0.30: return text # Необратимая потеря данных
original_score = text_quality_score(text)
best_text = text
best_score = original_score
if best_score > 2.5: return text # Уже читаемый
# FIRST: Handle common UTF-8 mojibake patterns (UTF-8 bytes decoded as Latin-1)
# This handles cases like 'Ð\x9fÑивеÑ\x82' which should be 'Привет'
# If the text contains common mojibake patterns, try to fix them immediately
if re.search(r'[ÐÑ][\x80-\x9f\xa0-\xbf]', text):
try:
utf8_fixed = text.encode('latin1', errors='ignore').decode('utf-8', errors='replace')
utf8_score = text_quality_score(utf8_fixed)
# If the UTF-8 fix improves the score significantly, return it immediately
if utf8_score > best_score and utf8_score > 0:
return utf8_fixed
# Otherwise, still consider it as a candidate
elif utf8_score > best_score:
best_score = utf8_score
best_text = utf8_fixed
except:
pass # Continue with other methods if UTF-8 fix fails
# Try to extract charset from HTML to handle your specific case
declared_charset = extract_html_charset(text)
raw_bytes = text.encode('latin1', errors='replace')
# Check if this looks like UTF-8 decoded Windows-1251 content with replacement chars ()
# If we have replacement characters, it indicates irreversible data loss during previous encoding
has_replacement_chars = '' in text
# If we have replacement characters, the data is already lost, so return original
if has_replacement_chars:
return text
# Prioritize the declared charset if found, especially for HTML content with windows-1251
base_encs = ['windows-1251', 'koi8-r', 'cp1251', 'utf-8', 'cp866', 'mac_cyrillic']
encs = [declared_charset] + [e for e in base_encs if e != declared_charset] if declared_charset else base_encs
for enc in encs:
for swap in [False, True]:
try:
decoded = raw_bytes.decode(enc, errors='replace')
if swap: decoded = decoded.swapcase()
score = text_quality_score(decoded)
# Берем вариант с максимальным скором (без жесткого порога)
if score > best_score:
best_score = score
best_text = decoded
except Exception:
continue
# Special handling for Windows-1251 mojibake specifically
# This targets the issue where Windows-1251 bytes were incorrectly decoded as UTF-8
try:
# If we have suspected mojibake text (many non-readable characters), try to recover
suspected_mojibake_score = text_quality_score(text)
# Attempt recovery from UTF-8 misinterpretation of Windows-1251 (before replacement chars appear)
try:
recovered_from_win1251 = raw_bytes.decode('windows-1251', errors='replace')
recovery_score = text_quality_score(recovered_from_win1251)
if recovery_score > best_score:
best_score = recovery_score
best_text = recovered_from_win1251
except Exception:
pass
except Exception:
pass
# Special handling for HTML content with charset declaration
if declared_charset and '1251' in declared_charset and (' best_score:
best_score = score
best_text = decoded_html
except:
continue
except:
pass # Continue with standard processing if special handling fails
# 🔁 Двойной mojibake
if re.search(r'[\u00C2-\u00C3][\u0080-\u00BF]', best_text):
try:
double_fixed = best_text.encode('latin1', errors='replace').decode('utf-8', errors='replace')
if text_quality_score(double_fixed) > best_score:
best_text = double_fixed
except Exception: pass
# Additional fix for common mojibake patterns that appear in your text
# Handle specific mojibake patterns like â, €,
try:
# Pattern for common mojibake sequences
parts = re.split(r'(â€[\x80-\xBF])|(â\x80[\x80-\x99])', best_text)
fixed_parts = []
for part in parts:
if part:
try:
# Try to fix common mojibake patterns
fixed_part = part.encode('latin1', errors='replace').decode('utf-8', errors='replace')
if text_quality_score(fixed_part) > text_quality_score(part):
fixed_parts.append(fixed_part)
else:
fixed_parts.append(part)
except:
fixed_parts.append(part)
best_text = ''.join(fixed_parts)
except Exception:
pass
# CRITICAL FIX: If the decoded text is worse than the original, return the original
decoded_score = text_quality_score(best_text)
if decoded_score < original_score - 0.5: # If decoded is significantly worse
return text
return best_text
def clean_html_aggressive(text):
text = re.sub(r']*>.*?', '', text, flags=re.DOTALL | re.IGNORECASE)
text = re.sub(r']*>.*?', '', text, flags=re.DOTALL | re.IGNORECASE)
# Remove HTML conditional comments like
text = re.sub(r'', '', text, flags=re.DOTALL | re.IGNORECASE)
# Also handle any remaining conditional comment start/end without proper pairing
text = re.sub(r'', '', text, flags=re.DOTALL | re.IGNORECASE)
# Remove regular HTML comments after handling conditional ones
text = re.sub(r'', '', text, flags=re.DOTALL | re.IGNORECASE)
text = re.sub(r'<[^>]+>', '', text)
text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)
return text
def process_text(raw_text, fix_encoding_flag=True, source_encoding='auto'):
if not isinstance(raw_text, str): raw_text = str(raw_text)
# Store the original text to return if decoding doesn't improve readability
original_raw_text = raw_text
# First, check if the original text has replacement characters or excessive question marks
# which indicates irreversible data loss
has_replacement_chars = '\ufffd' in raw_text or raw_text.count('?') / len(raw_text) > 0.2 if raw_text else False
cleaned = clean_html_aggressive(raw_text)
cleaned = html.unescape(cleaned)
if fix_encoding_flag and not has_replacement_chars:
# Only apply fix if there's no clear evidence of irreversible data loss
cleaned = fix_mojibake_by_decoding(cleaned)
elif has_replacement_chars:
# If we have replacement chars, return original to avoid degrading it further
return original_raw_text
cleaned = re.sub(r' +', ' ', cleaned).replace('\xa0', ' ')
lines = [line.strip() for line in cleaned.split('\n')]
result = '\n'.join(line for line in lines if line)
# Compare quality scores - return original if processing made it worse
if fix_encoding_flag and not has_replacement_chars:
original_score = text_quality_score(original_raw_text)
processed_score = text_quality_score(result)
# If the processed text is significantly worse than original, return original
if processed_score < original_score - 0.5:
return original_raw_text
return result
def decode_bytes_to_best_string(data_bytes):
best = data_bytes.decode('utf-8', errors='replace')
if count_cyrillic(best) < 3:
for enc in ['cp1251', 'cp866', 'koi8-r', 'latin1']:
try:
dec = data_bytes.decode(enc, errors='replace')
if count_cyrillic(dec) > count_cyrillic(best): best = dec
except: pass
return fix_mojibake_by_decoding(best)
best = data_bytes.decode('utf-8', errors='replace')
if count_cyrillic(best) < 3:
for enc in ['cp1251', 'cp866', 'koi8-r', 'latin1']:
try:
dec = data_bytes.decode(enc, errors='replace')
if count_cyrillic(dec) > count_cyrillic(best): best = dec
except: pass
return fix_mojibake_by_decoding(best)
def ensure_utf8_string(text: str) -> str:
return text.encode('utf-8', errors='replace').decode('utf-8')
# --------------------------------------------------------
class LocationHandler(BaseHTTPRequestHandler):
def send_json_error(self, code, message):
self.send_response(code)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps({'error': message}, ensure_ascii=False).encode('utf-8'))
def do_GET(self):
if len(self.path) > MAX_URI_LENGTH:
self.send_response(400)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.end_headers()
self.wfile.write(json.dumps({"error": "URI Too Long", "message": "Превышен лимит. Используйте POST."}, ensure_ascii=False).encode('utf-8'))
return
try:
parsed = urlparse(self.path)
path = unquote(parsed.path)
qp = parse_qs(parsed.query)
if path in ('/clean', '/text') or path.startswith('/clean/') or path.startswith('/text/'):
inp = path[7:] if path.startswith('/clean/') else (path[6:] if path.startswith('/text/') else qp.get('text', [''])[0])
if not inp: return self.send_json_error(400, "Отсутствует 'text'")
if len(inp) > MAX_GET_TEXT_LENGTH: return self.send_json_error(400, f"Макс. {MAX_GET_TEXT_LENGTH} для GET. Используйте POST.")
fix = qp.get('fix_encoding', ['true'])[0].lower() in ('true', '1', 'yes')
cleaned = process_text(inp, fix, qp.get('encoding', ['auto'])[0])
cleaned = ensure_utf8_string(cleaned)
self._send_json(200, {'cleaned_text': cleaned})
return
inp = path.lstrip('/')
if not inp:
if 'ip' in qp: inp, it = qp['ip'][0], "ip"
elif 'phone' in qp: inp, it = qp['phone'][0], "phone"
else: return self.send_json_error(400, "Требуется IP или телефон")
else: it = detect_input_type(inp)
res = get_region_by_ip(inp) if it == "ip" else (get_region_by_phone(inp) if it == "phone" else None)
if not res: return self.send_json_error(400, "Не удалось определить тип данных")
self._send_json(200, res)
except Exception as e:
self.send_json_error(500, f"Ошибка: {e}")
print(traceback.format_exc())
def do_POST(self):
try:
cl = int(self.headers.get('Content-Length', 0))
if cl > 10*1024*1024: return self.send_json_error(413, "Макс. 10МБ")
data = self.rfile.read(cl)
ct = self.headers.get('Content-Type', '').lower()
if not data:
qp = parse_qs(urlparse(self.path).query)
if 'text' in qp:
cleaned = process_text(qp['text'][0], qp.get('fix_encoding',['true'])[0].lower() in ('true','1'), qp.get('encoding',['auto'])[0])
return self._send_json(200, {'cleaned_text': ensure_utf8_string(cleaned)})
return self.send_json_error(400, "Пустое тело")
if 'json' in ct:
ch = ct.split('charset=')[1].split(';')[0].strip() if 'charset=' in ct else 'utf-8'
try: txt = data.decode(ch)
except: txt = data.decode('utf-8', errors='replace')
try: d = json.loads(txt)
except: return self.send_json_error(400, "Неверный JSON")
if 'text' in d:
cleaned = process_text(d['text'], d.get('fix_encoding', True), d.get('encoding', 'auto'))
return self._send_json(200, {'cleaned_text': ensure_utf8_string(cleaned)})
inp = d.get('ip') or d.get('phone') or d.get('query', '')
it = "ip" if d.get('ip') else ("phone" if d.get('phone') else detect_input_type(inp))
if not inp: return self.send_json_error(400, "Требуется IP/телефон")
res = get_region_by_ip(inp) if it=="ip" else get_region_by_phone(inp)
return self._send_json(200, res)
# Plain text / fallback
cleaned = process_text(decode_bytes_to_best_string(data), True, 'auto')
self._send_json(200, {'cleaned_text': ensure_utf8_string(cleaned)})
except Exception as e:
self.send_json_error(500, f"Ошибка: {e}")
print(traceback.format_exc())
def _send_json(self, code, obj):
self.send_response(code)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps(obj, ensure_ascii=False).encode('utf-8'))
def log_message(self, format, *args): pass
def decrypt_garbled_text(text: str) -> str:
"""Attempt to recover garbled UTF-8/CP1251 text.
Uses the existing :func:`fix_mojibake_by_decoding` and :func:`clean_html_aggressive`.
Returns the best human‑readable string or the original if no improvement.
"""
if not text:
return text
# Quick sanity: if already looks good, skip work
if text_quality_score(text) > 2.5 and count_cyrillic(text) > 0:
return text
# Remove HTML artifacts
cleaned = clean_html_aggressive(text)
cleaned = html.unescape(cleaned)
result = fix_mojibake_by_decoding(cleaned)
if count_cyrillic(result) == 0:
# fallback: try other decodings
result = process_text(text, True, 'auto')
return result
def run_server(port=8887):
load_russian_phone_database()
httpd = HTTPServer(('0.0.0.0', port), LocationHandler)
print(f" Запуск сервиса геолокации на порту {port}...")
print(" База данных IP загружена")
print(f" База телефонов: загружено {len(russian_phone_db) if russian_phone_db else 0} диапазонов")
print("\n Поддерживаемые эндпоинты:")
print(" GET / - определение местоположения по IP")
print(" GET /<номер_телефона> - определение местоположения по телефону")
print(" GET /?ip= - определение местоположения по IP (query-параметр)")
print(" GET /?phone=<номер_телефона> - определение местоположения по телефону (query-параметр)")
print(" POST / c JSON: {\"ip\": \"...\"} или {\"phone\": \"...\"}")
print(" POST / c JSON: {\"text\": \"...\", \"encoding\": \"auto\", \"fix_encoding\": true}")
print(" POST / с plain/text или text/html - автоматическая очистка и декодирование")
print(" Для больших текстов (>2000 символов) всегда используйте POST.")
print(" Все ответы строго в UTF-8 с заголовком Content-Type: application/json; charset=utf-8")
print(" Ошибки возвращаются в JSON, заголовки строго ASCII (совместимо с HTTP/1.1)")
print(" Автоматически извлекает declared charset, детектит потерю данных и использует частотный анализ букв")
print("\n Нажмите Ctrl+C для остановки сервера\n")
try: httpd.serve_forever()
except KeyboardInterrupt: print("\n Остановлен.")
finally: httpd.server_close()
if __name__ == '__main__':
run_server()