"""CarCompare website. / user UI: budget search + filters /car/ car ki full details /compare 2-4 cars side-by-side /admin admin panel (password protected): DB sync, cars manage Data sirf local DB (db.py) se aata hai, scraper.py use bharta hai. """ from datetime import datetime from functools import wraps import hmac import json import os import secrets import threading from contextlib import closing from flask import (Flask, Response, flash, redirect, render_template, request, url_for, session, abort) import db import scraper import catalog import ingestion import blog_writer app = Flask(__name__) app.secret_key = os.environ.get('SECRET_KEY') or os.urandom(24) app.config.update(MAX_CONTENT_LENGTH=8 * 1024 * 1024, SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE='Lax') # Small, server-side UI dictionary. Vehicle/blog data remains in its original # language; this setting translates the shared navigation and labels. UI_TEXT = { 'en': {'find_cars':'Find Cars','find_bikes':'Find Bikes','compare':'Compare','journal':'The Journal', 'clear':'Clear','explore':'Explore','blog':'Blog','popular_budgets':'Popular Budgets', 'language':'Language','english':'English','hindi':'हिन्दी','compare_arrow':'Compare →'}, 'hi': {'find_cars':'कार खोजें','find_bikes':'बाइक खोजें','compare':'तुलना करें','journal':'जर्नल', 'clear':'साफ़ करें','explore':'ब्राउज़ करें','blog':'ब्लॉग','popular_budgets':'लोकप्रिय बजट', 'language':'भाषा','english':'English','hindi':'हिन्दी','compare_arrow':'तुलना करें →'}, } @app.get('/language/') def set_language(code): if code in UI_TEXT: session['language'] = code return redirect(request.referrer or url_for('home')) # Admin login - production me env variables zaroor set karein ADMIN_USER = os.environ.get('ADMIN_USER', 'admin') ADMIN_PASSWORD = os.environ.get('ADMIN_PASSWORD', 'admin123') @app.context_processor def inject_globals(): with closing(db.connect()) as conn: media = dict(conn.execute('SELECT source_url,local_path FROM media_cache')) language = session.get('language', 'en') return {'current_year': datetime.now().year, 'site': catalog.content(), 'language': language, 'ui': UI_TEXT.get(language, UI_TEXT['en']), 'local_image': lambda value: value if value and value.startswith('/static/media/') else media.get(value, url_for('static',filename='vehicle-placeholder.svg')), 'csrf_token': csrf_token} def csrf_token(): if '_csrf' not in session: session['_csrf'] = secrets.token_urlsafe(32) return session['_csrf'] @app.before_request def protect_admin_forms(): if request.method == 'POST' and request.path.startswith('/admin'): expected = session.get('_csrf', '') if not expected or not hmac.compare_digest(expected, request.form.get('_csrf','')): abort(400, 'Form expired. Reload the page and try again.') SORT_LABELS = [ ('relevance', 'Budget ke paas'), ('price_asc', 'Price: kam se zyada'), ('price_desc', 'Price: zyada se kam'), ('rating', 'User rating'), ('mileage', 'Mileage'), ] # --------------------------------------------------------- # USER UI # --------------------------------------------------------- @app.route('/') def home(): budget = request.args.get('budget', type=float) tolerance = request.args.get('range', default=1.0, type=float) sort = request.args.get('sort', 'relevance') selected = { 'brands': request.args.getlist('brand'), 'bodies': request.args.getlist('body'), 'fuels': request.args.getlist('fuel'), 'transmissions': request.args.getlist('trans'), } cars = db.search_cars(budget, tolerance, sort=sort, **selected) by_brand = {} for car in cars: by_brand[car['brand']] = by_brand.get(car['brand'], 0) + 1 picks = db.top_rated_cars(4) chosen = catalog.content().get('spotlight_id') spotlight = db.get_car(chosen)[0] if chosen else None return render_template( 'index.html', cars=cars, by_brand=list(by_brand.items()), budget=budget, tolerance=tolerance, sort=sort, sorts=SORT_LABELS, sel=selected, opts=db.filter_options(), total_cars=db.stats()['cars'], latest_posts=db.list_posts(published_only=True, limit=3), top_cars=picks, spotlight=spotlight or (picks[0] if picks else None), body_types=db.body_type_stats(), brands=db.brand_stats(), is_default_view=not (budget or any(selected.values())), ) @app.route('/car/') def car_detail(model_no): car, details = db.get_car(model_no) if not car: return redirect(url_for('home')) similar = [] if car['min_price']: similar = [ c for c in db.search_cars(car['min_price'], 1.5) if c['model_no'] != model_no ][:4] return render_template('car.html', car=car, d=details, similar=similar) @app.route('/blog') def blog_list(): tag = request.args.get('tag') posts = db.list_posts(published_only=True) if tag: posts = [p for p in posts if p['tag'] == tag] tags = sorted({p['tag'] for p in db.list_posts(published_only=True) if p['tag']}) return render_template('blog_list.html', posts=posts, tags=tags, active_tag=tag) @app.route('/bikes') def bike_list(): category = request.args.get('category', '') query = request.args.get('q', '').strip() categories = sorted({bike['category'] for bike in catalog.list_bikes() if bike['category']}) bikes = catalog.list_bikes(query,category) return render_template('bikes.html', bikes=bikes, categories=categories, category=category, query=query) @app.route('/bike/') def bike_detail(model_no): bike = catalog.get_bike(model_no) if not bike: abort(404) return render_template('bike.html', bike=bike) @app.route('/blog/') def blog_post(slug): post = db.get_post(slug) if not post or not post['published']: return redirect(url_for('blog_list')) related = [p for p in db.list_posts(published_only=True, exclude_id=post['id']) if p['tag'] == post['tag']][:3] if len(related) < 3: others = [p for p in db.list_posts(published_only=True, exclude_id=post['id']) if p not in related][:3 - len(related)] related += others return render_template('blog_post.html', post=post, related=related) @app.route('/compare') def compare(): model_nos = [m for m in request.args.get('cars', '').split(',') if m] compare_type = request.args.get('type', 'car') if compare_type == 'bike': cars, rows = catalog.compare_bikes(model_nos) features = {} tray = [{'id': c['model_no'], 'name': c['name'], 'kind': 'bike'} for c in cars] else: cars, rows, features = db.compare_cars(model_nos) tray = [{'id': c['model_no'], 'name': c['car_name'], 'kind': 'car'} for c in cars] return render_template( 'compare.html', cars=cars, rows=rows, features=features, model_nos=[c['model_no'] for c in cars], compare_type=compare_type, # URL me cars hon to browser ki compare list usi se sync ho tray_json=json.dumps(tray) if 'cars' in request.args else '', ) # --------------------------------------------------------- # ADMIN PANEL # --------------------------------------------------------- def admin_required(view): @wraps(view) def wrapper(*args, **kwargs): auth = request.authorization if not session.get('admin') and not ( auth and hmac.compare_digest(auth.username or '', ADMIN_USER) and hmac.compare_digest(auth.password or '', ADMIN_PASSWORD) ): return redirect(url_for('admin_login')) return view(*args, **kwargs) return wrapper @app.route('/admin') @admin_required def admin(): return render_template( 'admin.html', stats=db.stats(), brands=db.brand_stats(), cars=db.admin_car_list(), last_sync=db.last_sync_time(), db_file=db.DB_FILE, posts=db.list_posts(published_only=False), bikes=catalog.list_bikes(), sources=catalog.sources(), blog_sources=catalog.blog_sources(), runs=catalog.runs(), content_fields=catalog.CONTENT, tab=request.args.get('tab','overview'), blog_jobs=blog_writer.jobs(), blog_writer_ready=blog_writer.configured(), ) @app.route('/admin/login', methods=['GET','POST']) def admin_login(): if request.method == 'POST': if hmac.compare_digest(request.form.get('username',''),ADMIN_USER) and hmac.compare_digest(request.form.get('password',''),ADMIN_PASSWORD): session.clear(); session['admin']=True return redirect(url_for('admin')) flash('Incorrect username or password.','error') return render_template('admin_login.html') @app.post('/admin/logout') def admin_logout(): session.clear() return redirect(url_for('admin_login')) @app.post('/admin/content') @admin_required def admin_content(): if not request.form.get('site_name','').strip(): abort(400,'Site name is required.') spotlight=request.form.get('spotlight_id','').strip() if spotlight and not db.get_car(spotlight)[0]: abort(400,'Spotlight car ID does not exist.') catalog.save_content(request.form) flash('Site content saved. Public pages now use these values.') return redirect(url_for('admin',tab='content')) @app.post('/admin/sources') @admin_required def admin_source_save(): provider=request.form.get('provider',''); kind=request.form.get('kind','') url=request.form.get('url','').strip(); brand=request.form.get('brand','').strip() try: if kind == 'blog': ingestion.validate_blog_source(provider,url) with closing(db.connect()) as conn,conn: conn.execute('INSERT INTO blog_sources(provider,url,tag) VALUES (?,?,?)', (provider,url,brand or 'Imported')) flash('Blog feed added. Run a blog sync to import draft stories.') else: ingestion.validate_source(provider,kind,url) if not brand: raise ValueError('Brand is required.') with closing(db.connect()) as conn,conn: conn.execute('INSERT INTO scrape_sources(kind,brand,provider,url) VALUES (?,?,?,?)',(kind,brand,provider,url)) flash('Source added. Run a sync to import its models.') except (ValueError,db.sqlite3.IntegrityError) as error: flash(str(error),'error') return redirect(url_for('admin',tab='sources')) @app.post('/admin/sources//toggle') @admin_required def admin_source_toggle(source_id): with closing(db.connect()) as conn,conn: conn.execute('UPDATE scrape_sources SET enabled=1-enabled WHERE id=?',(source_id,)) return redirect(url_for('admin',tab='sources')) @app.post('/admin/blog-sources//toggle') @admin_required def admin_blog_source_toggle(source_id): with closing(db.connect()) as conn,conn: conn.execute('UPDATE blog_sources SET enabled=1-enabled WHERE id=?',(source_id,)) return redirect(url_for('admin',tab='sources')) @app.route('/admin/vehicles///edit',methods=['GET','POST']) @admin_required def admin_vehicle_edit(kind,model_no): if kind not in ('car','bike'): abort(404) vehicle=db.get_car(model_no)[0] if kind=='car' else catalog.get_bike(model_no) if not vehicle: abort(404) fields=['car_name','price','min_price','max_price','body_type','fuel_type','mileage','engine','seating','transmission','description'] if kind=='car' else ['name','price','min_price','max_price','category','description'] if request.method=='POST': try: values={key:request.form.get(key,'').strip() for key in fields} if not values.get('car_name',values.get('name','')): raise ValueError('Name is required.') for key in ('min_price','max_price'): values[key]=float(values[key]) if values[key] else None if values[key] is not None and (not __import__('math').isfinite(values[key]) or values[key]<0): raise ValueError('Prices must be finite and non-negative.') if values['min_price'] is not None and values['max_price'] is not None and values['min_price']>values['max_price']: raise ValueError('Maximum price must not be below minimum price.') with closing(db.connect()) as conn,conn: table='cars' if kind=='car' else 'bikes' conn.execute(f'UPDATE {table} SET '+','.join(key+'=?' for key in fields)+' WHERE model_no=?',[*values.values(),model_no]) flash('Vehicle content saved. A future sync may replace scraped fields.') return redirect(url_for('admin',tab='cars' if kind=='car' else 'bikes')) except ValueError as error: flash(str(error),'error'); vehicle.update(request.form) return render_template('admin_vehicle_form.html',vehicle=vehicle,fields=fields,kind=kind) @app.post('/admin/bike//delete') @admin_required def admin_delete_bike(model_no): with closing(db.connect()) as conn,conn: conn.execute('DELETE FROM bikes WHERE model_no=?',(model_no,)) flash('Bike removed. An enabled source can import it again on the next sync.') return redirect(url_for('admin',tab='bikes')) @app.route('/admin/posts/new', methods=['GET', 'POST']) @admin_required def admin_post_new(): return admin_post_form(None) @app.post('/admin/posts/generate') @admin_required def admin_post_generate(): try: job_id = blog_writer.enqueue(request.form.get('topic', '')) except ValueError as error: flash(str(error), 'error') return redirect(url_for('admin', tab='blog')) try: threading.Thread(target=blog_writer.run, args=(job_id,), daemon=True).start() except RuntimeError: with closing(db.connect()) as conn, conn: conn.execute("UPDATE blog_jobs SET status='failed',error='Could not start the writer. Please retry.' WHERE id=?", (job_id,)) flash('Could not start the writer. Please retry.', 'error') else: flash('Research started. Your article will appear here as an editable draft.') return redirect(url_for('admin', tab='blog')) @app.get('/admin/posts/generation-status') @admin_required def admin_generation_status(): return {'jobs': [{key: job[key] for key in ('id','topic','status','post_id','error')} for job in blog_writer.jobs()]} @app.route('/admin/posts//edit', methods=['GET', 'POST']) @admin_required def admin_post_edit(post_id): return admin_post_form(post_id) def admin_post_form(post_id): post = db.get_post_by_id(post_id) if post_id else None if post_id and not post: return redirect(url_for('admin') + '#blog') if request.method == 'POST': title = request.form.get('title', '').strip() if not title: flash('Title zaroori hai.', 'error') else: cover_image=post['cover_image'] if post else '' upload=request.files.get('cover_upload') if upload and upload.filename: from PIL import Image from io import BytesIO import hashlib try: data=upload.read() with Image.open(BytesIO(data)) as picture: picture.verify() extension={'JPEG':'.jpg','PNG':'.png','WEBP':'.webp'}.get(picture.format) if not extension: raise ValueError('Use a JPG, PNG or WebP image.') filename=hashlib.sha256(data).hexdigest()+extension ingestion.MEDIA_DIR.mkdir(exist_ok=True) (ingestion.MEDIA_DIR/filename).write_bytes(data) cover_image='/static/media/'+filename except Exception: flash('Cover upload is not a valid JPG, PNG or WebP image.','error') return render_template('admin_post_form.html',post={**(post or {}),**request.form,'published':bool(request.form.get('published'))}),400 slug = db.save_post( post_id, title=title, excerpt=request.form.get('excerpt', '').strip(), content=request.form.get('content', '').strip(), cover_image=cover_image, author=request.form.get('author', '').strip() or 'CarCompare Team', tag=request.form.get('tag', '').strip(), published=1 if request.form.get('published') else 0, ) flash(f'Post "{title}" saved.') return redirect(url_for('blog_post', slug=slug) if request.form.get('published') else url_for('admin',tab='blog')) return render_template('admin_post_form.html', post=post) @app.route('/admin/posts//publish', methods=['POST']) @admin_required def admin_post_publish(post_id): db.toggle_publish(post_id) return redirect(url_for('admin',tab='blog')) @app.route('/admin/posts//delete', methods=['POST']) @admin_required def admin_post_delete(post_id): db.delete_post(post_id) flash('Post deleted.') return redirect(url_for('admin',tab='blog')) @app.route('/admin/sync', methods=['POST']) @admin_required def admin_sync(): mode = request.form.get('mode', 'full') kind = request.form.get('kind','all') if mode not in ('full','missing') or kind not in ('car','bike','blog','all'): abort(400) run_id=catalog.start_run(kind,mode) if run_id: threading.Thread(target=ingestion.run_sync,args=(kind,mode,run_id),daemon=True).start() flash('Sync started. You can keep working while the catalogue updates.') else: flash('A sync is already active. Check its status below.','error') return redirect(url_for('admin',tab='sync')) @app.get('/admin/sync/status') @admin_required def admin_sync_status(): return {'runs':catalog.runs()} @app.route('/admin/car//refresh', methods=['POST']) @admin_required def admin_refresh_car(model_no): car, _ = db.get_car(model_no) if car: try: if 'carwale.com' in (car['source_url'] or ''): source=next(s for s in catalog.sources() if s['brand']==car['brand'] and s['provider']=='carwale') result=ingestion.parse_model(ingestion.fetch_state(car['source_url']),source,car['source_url']) ingestion.save_model(result,source) ingestion.cache_image(result['image']) else: done, errors=scraper.scrape_details([car]) if errors: raise ValueError('; '.join(errors)) flash(f"{car['car_name']} refreshed.") except Exception as error: flash(f'Refresh failed; saved data retained. {error}','error') return redirect(url_for('admin',tab='cars')) @app.route('/admin/car//delete', methods=['POST']) @admin_required def admin_delete_car(model_no): db.delete_car(model_no) flash(f'{model_no} deleted.') return redirect(url_for('admin',tab='cars')) db.init_db() catalog.init() if __name__ == '__main__': app.run(debug=True)