# Libraries from flask import Flask, render_template, request, send_file, flash, redirect, url_for, session from flask_httpauth import HTTPBasicAuth from flask_session import Session import zipfile from io import BytesIO from datetime import timedelta, datetime import asyncio from telegram import Bot, error from tenacity import retry, stop_after_attempt, wait_exponential from werkzeug.security import generate_password_hash, check_password_hash # Init App app = Flask(__name__) UPLOAD_FOLDER = 'uploads' app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER app.config['SESSION_TYPE'] = 'filesystem' app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(minutes=10) app.config['SESSION_FILE_DIR'] = '/var/www/cvtools/html/flask_session_cache' Session(app) # Telegram Bot Token TELEGRAM_BOT_TOKEN = '7011981221:AAE_XUX7x9XcVMyFkkgKdd59Ff9NsNKxQPs' bot = Bot(token=TELEGRAM_BOT_TOKEN) app.secret_key = 'KontolLOdonAnjayaysusugusuheuh' auth = HTTPBasicAuth() # users = { # "karyaagung": generate_password_hash("Fakhri11072008!"), # } def authenticate(username, password): users = { "kripermana": generate_password_hash("Fakhri11072008!"), "ssaprhn": generate_password_hash("parhanganteng15"), "nursaffaanah_": generate_password_hash("Saffa26") } return username in users and check_password_hash(users[username], password) @app.before_request def before_request(): session.permanent = True session.modified = True now = datetime.now() if 'last_activity' in session: last_activity = session['last_activity'] if (now- last_activity).total_seconds() > app.config['PERMANENT_SESSION_LIFETIME'].total_seconds(): session.clear() return redirect(url_for('login')) session['last_activity'] = now @app.route('/login', methods=['GET', 'POST']) def login(): if request.method == 'POST': username = request.form['username'] password = request.form['password'] if authenticate(username, password): session['username'] = username session['last_activity'] = datetime.now() return redirect(url_for('index')) else: flash('Invalid username or password') return render_template('login.html') @app.route('/logout') def logout(): session.clear() return redirect(url_for('login')) user_chat_ids = { 'kripermana': 6272083115, 'pierdecp': 5980655096 } # Define template VCARD def create_vcf_contact(name, phone_number): return f"BEGIN:VCARD\nVERSION:3.0\nFN:{name}\nTEL;TYPE=CELL:{phone_number}\nEND:VCARD\n\n" # Create VCF File def create_vcf_file(admins, navies, clients, action, max_contacts_per_file, file_name, include_admin_navy, admin_contact_name_prefix, navy_contact_name_prefix, client_contact_name_prefix, startFrom): files = {} vcf_content = "" if action == "1": # Gabung Admin, Navy, Client for i, admin in enumerate(admins): if not admin.startswith('+'): admin = '+' + admin vcf_content += create_vcf_contact(f"{admin_contact_name_prefix}{i+1}", admin) vcf_content += "\n" for i, navy in enumerate(navies): if not navy.startswith('+'): navy = '+' + navy vcf_content += create_vcf_contact(f"{navy_contact_name_prefix}{i+1}", navy) vcf_content += "\n" for i, client in enumerate(clients): if not client.startswith('+'): client = '+' + client vcf_content += create_vcf_contact(f"{client_contact_name_prefix}{i+1}", client) files[f"{file_name}.vcf"] = vcf_content else: # Pecah file if include_admin_navy: admin_navy_vcf = "" for i, admin in enumerate(admins): if not admin.startswith('+'): admin = '+' + admin admin_navy_vcf += create_vcf_contact(f"{admin_contact_name_prefix}{i+1}", admin) admin_navy_vcf += "\n" for i, navy in enumerate(navies): if not navy.startswith('+'): navy = '+' + navy admin_navy_vcf += create_vcf_contact(f"{navy_contact_name_prefix}{i+1}", navy) admin_navy_vcf += "\n" else: admin_navy_vcf = "" # Buat file admin dan navy admin_navy = "" for i, admin in enumerate(admins): if not admin.startswith('+'): admin = '+' + admin admin_navy += create_vcf_contact(f"{admin_contact_name_prefix}{i+1}", admin) admin_navy += "\n" for i, navy in enumerate(navies): if not navy.startswith('+'): navy = '+' + navy admin_navy += create_vcf_contact(f"{navy_contact_name_prefix}{i+1}", navy) files[f"{file_name}admin.vcf"] = admin_navy for i in range(0, len(clients), max_contacts_per_file): client_vcf = admin_navy_vcf if include_admin_navy else "" for j, client in enumerate(clients[i:i + max_contacts_per_file]): if not client.startswith('+'): client = '+' + client client_vcf += create_vcf_contact(f"{client_contact_name_prefix}{i + j + 1}", client) files[f"{file_name}{i //max_contacts_per_file + startFrom}.vcf"] = client_vcf return files async def send_via_telegram(username, files): bot = Bot(token=TELEGRAM_BOT_TOKEN) chat_id = user_chat_ids.get(username.lstrip('@')) if not chat_id: print(f"Chat ID buat {username} gak ketemu boss...") for file_name, content in files.items(): print(f"sedang mengirim {file_name} ke tujuan username {username}") try: await bot.send_document(chat_id=chat_id, document=BytesIO(content.encode()), filename=file_name, write_timeout=25) except error.TimedOut: print(f"Timeout saat mengirim {file_name} ke tujuan username {username}") # Retry sending the remaining files for remaining_file_name, remaining_content in list(files.items())[list(files.keys()).index(file_name) + 1:]: try: await bot.send_document(chat_id=chat_id, document=BytesIO(remaining_content.encode()), filename=remaining_file_name, write_timeout=25) except error.TimedOut: print(f"Timeout saat mengirim {remaining_file_name} ke tujuan username {username}") @app.route('/', methods=['GET', 'POST']) def index(): if 'username' not in session: return redirect(url_for('login')) if request.method == 'POST': admin_numbers = request.form['admin_numbers'].strip().split('\n') navy_numbers = request.form['navy_numbers'].strip().split('\n') client_numbers = request.form['client_numbers'].strip().split('\n') action = request.form['opsi-export'] include_admin_navy = 'include_admin_navy' in request.form file_name = request.form['file_name'] or f"contacts_{datetime.datetime.now().strftime('%d%m%Y_%H%M%S')}" max_contacts_per_file = int(request.form['max_contacts_per_file']) if request.form['max_contacts_per_file'] else len(client_numbers) startFrom = int(request.form['startFrom']) if request.form['startFrom'] else 1 admin_contact_name_prefix = request.form['admin_contact_name_prefix'] or "ADMIN" navy_contact_name_prefix = request.form['navy_contact_name_prefix'] or "NAVY" client_contact_name_prefix = request.form['client_contact_name_prefix'] or "CLIENT" telegram_username = request.form['telegram_username'] download_type = request.form['download_type'] # Hapus line yang kosong admin_numbers = [number.strip() for number in admin_numbers if number.strip()] navy_numbers = [number.strip() for number in navy_numbers if number.strip()] client_numbers = [number.strip() for number in client_numbers if number.strip()] # Generate file kontak files = create_vcf_file( admin_numbers, navy_numbers, client_numbers, action, max_contacts_per_file, file_name, include_admin_navy, admin_contact_name_prefix, navy_contact_name_prefix, client_contact_name_prefix, startFrom ) if telegram_username: asyncio.run(send_via_telegram(telegram_username, files)) flash(f"Kontak berhasil terkirim melalui telegram ke username {telegram_username}", "success") return redirect(url_for('index')) if download_type == "zip": zip_buffer = BytesIO() with zipfile.ZipFile(zip_buffer, 'w') as zip_file: for name, content in files.items(): zip_file.writestr(name, content) zip_buffer.seek(0) return send_file(zip_buffer, as_attachment=True, download_name=f"{file_name}.zip", mimetype='application/zip') else: if len(files) == 1: output_name, output_content = list(files.items())[0] return send_file(BytesIO(output_content.encode()), mimetype="text/vcard", download_name=output_name, as_attachment=True) else: zip_buffer = BytesIO() with zipfile.ZipFile(zip_buffer, 'w') as zf: for name, content in files.items(): zf.writestr(name, content) zip_buffer.seek(0) return send_file(zip_buffer, mimetype="application/zip", download_name=f"{file_name}.zip", as_attachment=True) return render_template('index.html') # Run if __name__ == '__main__': app.run(debug=True)