""" Import data from legacy PEZAJOB SQLite (store/profile_store schema) into upgraded models. Usage: python manage.py import_legacy_db D:\\db.sqlite3 python manage.py import_legacy_db D:\\db.sqlite3 --clear """ import re import sqlite3 from datetime import datetime from django.conf import settings from django.core.management.base import BaseCommand, CommandError from django.db import transaction from django.db.models import Count from django.db.models.signals import post_save from django.utils import timezone from django.utils.dateparse import parse_date, parse_datetime from django.utils.html import strip_tags from django.utils.text import slugify from pathlib import Path from accounts.models import User, UserNotification from companies.models import Company from home.models import DailyVisitorCount, SiteContact from jobs.models import Job, JobApplication, JobCategory, JobView from jobs.signals import on_job_posted from podcast.models import PodcastCategory from seekers.models import ( PortfolioStat, ProfileMessage, ProfileMessageReply, ProfileView, SeekerInterest, SeekerProfile, ) from subscriptions.models import ( CompanySubscription, PaymentMethod, SeekerSubscription, SubscriptionPayment, SubscriptionPlan, ) from subscriptions.benefits import sync_company_subscription_benefits, sync_seeker_subscription_benefits from home.cover_images import legacy_media_url ROLE_MAP = { 'job_seeker': User.Role.SEEKER, 'employer': User.Role.COMPANY, } EXPERIENCE_MAP = { '0-2': Job.ExperienceLevel.ENTRY, '3-5': Job.ExperienceLevel.MID, '6-10': Job.ExperienceLevel.SENIOR, '11-above': Job.ExperienceLevel.EXECUTIVE, } JOB_TYPE_MAP = { 'full-time': Job.JobType.FULL_TIME, 'full time': Job.JobType.FULL_TIME, 'part-time': Job.JobType.PART_TIME, 'part time': Job.JobType.PART_TIME, 'contract': Job.JobType.CONTRACT, 'remote': Job.JobType.REMOTE, 'internship': Job.JobType.INTERNSHIP, } APPLICATION_STATUS_MAP = { 'pending': JobApplication.Status.PENDING, 'shortlisted': JobApplication.Status.SHORTLISTED, 'hired': JobApplication.Status.HIRED, 'rejected': JobApplication.Status.REJECTED, } COMPANY_PLAN_BY_PERIOD = { 'Month': 'company-professional', 'Quarter': 'company-recruiter-pro', 'Year': 'company-enterprise', } COMPANY_TIER_BY_SLUG = { 'company-starter': Company.Tier.STARTER, 'company-professional': Company.Tier.PRO, 'company-recruiter-pro': Company.Tier.PREMIUM, 'company-enterprise': Company.Tier.ENTERPRISE, } DEFAULT_PAYMENT_METHODS = [ {'slug': 'mobile_money', 'name': 'Mobile Money', 'description': 'Airtel Money, MTN MoMo, Zamtel Kwacha', 'sort_order': 1}, {'slug': 'card', 'name': 'Debit / Credit Card', 'description': 'Visa and Mastercard via DPO', 'sort_order': 2}, {'slug': 'dpo', 'name': 'DPO Pay', 'description': '3G Direct Pay online checkout', 'sort_order': 3}, {'slug': 'paypal', 'name': 'PayPal', 'description': 'PayPal international payments', 'sort_order': 4}, ] LEGACY_PAYMENT_STATUS = { 'pending': SubscriptionPayment.Status.PENDING, 'paid': SubscriptionPayment.Status.PAID, 'delivered': SubscriptionPayment.Status.DELIVERED, 'rejected': SubscriptionPayment.Status.REJECTED, } def _clean(text): if text is None: return '' return str(text).strip() def _parse_dt(value): if not value: return timezone.now() if isinstance(value, datetime): dt = value else: dt = parse_datetime(str(value)) if dt is None: d = parse_date(str(value)) if d: dt = datetime.combine(d, datetime.min.time()) else: return timezone.now() if timezone.is_naive(dt): return timezone.make_aware(dt, timezone.get_current_timezone()) return dt def _parse_expiry(value): if not value: return None d = parse_date(str(value)) if not d: return None dt = datetime.combine(d, datetime.max.time().replace(microsecond=0)) return timezone.make_aware(dt, timezone.get_current_timezone()) def _safe_ip(value): ip = _clean(value) return ip[:45] if ip else '' def _unique_slug(model, base, pk=None): base = slugify(base) or 'item' slug = base counter = 1 while model.objects.filter(slug=slug).exclude(pk=pk).exists(): slug = f'{base}-{counter}' counter += 1 return slug def _infer_payment_method_slug(txn, provider_txn): txn_l = _clean(txn).lower() provider_l = _clean(provider_txn).lower() if 'paypal' in txn_l or 'paypal' in provider_l: return 'paypal' if provider_l.startswith('dpo_') or txn_l.startswith('sub-'): return 'dpo' if 'momo' in txn_l or 'mobile' in provider_l: return 'mobile_money' if 'card' in txn_l or 'card' in provider_l: return 'card' return 'dpo' def _parse_salary(salary_range): if not salary_range: return None, None nums = [int(n.replace(',', '')) for n in re.findall(r'\d+', str(salary_range))] if not nums: return None, None if len(nums) == 1: return nums[0], None return nums[0], nums[1] class Command(BaseCommand): help = 'Import users, companies, jobs, profiles, and applications from legacy db.sqlite3' def add_arguments(self, parser): parser.add_argument( 'source', nargs='?', default=r'c:\Users\tesla\OneDrive\Desktop\P\.idea 5\db.sqlite3', help='Path to legacy SQLite database', ) parser.add_argument( '--media-source', default='', help='Optional path to legacy media/ folder to copy into this project', ) parser.add_argument( '--clear', action='store_true', help='Delete existing imported data before import', ) def handle(self, *args, **options): source = Path(options['source']) if not source.exists(): raise CommandError(f'Legacy database not found: {source}') legacy = sqlite3.connect(source) legacy.row_factory = sqlite3.Row if options['clear']: self.stdout.write('Clearing current database content...') self._clear_data() post_save.disconnect(on_job_posted, sender=Job) stats = {} try: with transaction.atomic(): stats['users'] = self._import_users(legacy) stats['companies'] = self._import_companies(legacy) stats['categories'] = self._import_categories(legacy) stats['jobs'] = self._import_jobs(legacy) stats['profiles'] = self._import_profiles(legacy) stats['applications'] = self._import_applications(legacy) seeker_subs, company_subs = self._import_subscriptions(legacy) stats['seeker_subscriptions'] = seeker_subs stats['company_subscriptions'] = company_subs stats['testimonials'] = self._import_testimonials(legacy) stats['interests'] = self._import_interests(legacy) stats['portfolio_stats'] = self._import_portfolio_stats(legacy) stats['profile_views'] = self._import_profile_views(legacy) stats['profile_messages'] = self._import_profile_messages(legacy) stats['profile_replies'] = self._import_profile_replies(legacy) stats['notifications'] = self._import_notifications(legacy) stats['site_contacts'] = self._import_site_contacts(legacy) stats['visitor_days'] = self._import_visitor_traffic(legacy) stats['podcast_categories'] = self._import_podcast_categories(legacy) stats['job_views'] = self._import_job_views(legacy) stats['payment_methods'] = self._import_payment_methods(legacy) stats['subscription_payments'] = self._import_subscription_payments(legacy) self._sync_subscription_benefits() finally: post_save.connect(on_job_posted, sender=Job) legacy.close() if options.get('media_source'): stats['media_files'] = self._copy_media(options['media_source']) self.stdout.write(self.style.SUCCESS('Legacy import complete:')) for key, count in stats.items(): self.stdout.write(f' {key}: {count}') def _clear_data(self): JobView.objects.all().delete() JobApplication.objects.all().delete() Job.objects.all().delete() JobCategory.objects.all().delete() ProfileMessageReply.objects.all().delete() ProfileMessage.objects.all().delete() ProfileView.objects.all().delete() SeekerInterest.objects.all().delete() PortfolioStat.objects.all().delete() UserNotification.objects.all().delete() SiteContact.objects.all().delete() DailyVisitorCount.objects.all().delete() PodcastCategory.objects.all().delete() CompanySubscription.objects.all().delete() SeekerSubscription.objects.all().delete() SubscriptionPayment.objects.all().delete() PaymentMethod.objects.all().delete() SeekerProfile.objects.all().delete() Company.objects.all().delete() Testimonial.objects.all().delete() User.objects.filter(is_superuser=False).delete() def _import_users(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM accounts_user ORDER BY id') count = 0 for row in cur.fetchall(): email = _clean(row['email']).lower() if not email: continue username = _clean(row['username']) or email.split('@')[0] base_username = username suffix = 1 while User.objects.filter(username=username).exclude(pk=row['id']).exists(): username = f'{base_username}{suffix}' suffix += 1 role = ROLE_MAP.get(_clean(row['account_type']), User.Role.SEEKER) if row['is_superuser']: role = User.Role.ADMIN phone = row['phone'] phone_str = '' if phone in (None, 0) else str(phone) user, created = User.objects.update_or_create( pk=row['id'], defaults={ 'username': username, 'email': email, 'first_name': _clean(row['first_name']), 'last_name': _clean(row['last_name']), 'role': role, 'phone': phone_str, 'is_active': bool(row['is_active']), 'is_staff': bool(row['is_staff']), 'is_superuser': bool(row['is_superuser']), 'last_login': row['last_login'], 'date_joined': row['date_joined'] or datetime.now(), }, ) if created or user.password != row['password']: User.objects.filter(pk=user.pk).update(password=row['password']) count += 1 return count def _import_companies(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM business_businessaccount ORDER BY id') count = 0 for row in cur.fetchall(): owner_id = row['user_id'] if not owner_id or not User.objects.filter(pk=owner_id).exists(): continue name = _clean(row['business_name']) or _clean(row['legal_name']) or f'Company {row["id"]}' location = _clean(row['city_town']) or _clean(row['province']) or 'Zambia' slug = _unique_slug(Company, name, pk=row['id']) initials = ''.join(part[0] for part in name.split()[:2]).upper() or 'CO' cover_url = legacy_media_url(row['logo'], settings.MEDIA_URL) Company.objects.update_or_create( pk=row['id'], defaults={ 'owner_id': owner_id, 'name': name, 'slug': slug, 'initials': initials[:4], 'industry': _clean(row['industry']) or 'General', 'industry_slug': slugify(_clean(row['industry']) or 'general'), 'location': location, 'description': _clean(row['description']) or _clean(row['about_text']) or name, 'website': _clean(row['website_url']), 'email': _clean(row['business_email']) or f'company{row["id"]}@import.local', 'phone': _clean(row['business_phone_number_1']), 'cover_image_url': cover_url, 'is_active': _clean(row['status']) != 'rejected', 'is_featured': _clean(row['status']) == 'approved', }, ) count += 1 return count def _import_categories(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM store_itemcategories ORDER BY id') count = 0 for row in cur.fetchall(): name = _clean(row['name']) or f'Category {row["id"]}' slug = _clean(row['slug']) or _unique_slug(JobCategory, name, pk=row['id']) JobCategory.objects.update_or_create( pk=row['id'], defaults={'name': name, 'slug': slug}, ) count += 1 return count def _import_jobs(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM store_item ORDER BY id') count = 0 for row in cur.fetchall(): title = _clean(row['name']) or f'Job {row["id"]}' slug = _clean(row['slug']) or _unique_slug(Job, title, pk=row['id']) company_name = _clean(row['company_name']) posted_by_id = None if row['business_owner_id']: company = Company.objects.filter(pk=row['business_owner_id']).first() if company: company_name = company_name or company.name posted_by_id = company.owner_id if not company_name: company_name = 'Employer' location = _clean(row['city_town']) or _clean(row['quick_location']) or _clean(row['country']) or 'Zambia' job_type_key = _clean(row['job_type'] or row['quick_job_type']).lower() job_type = JOB_TYPE_MAP.get(job_type_key, Job.JobType.FULL_TIME) experience = EXPERIENCE_MAP.get(_clean(row['work_experience']), Job.ExperienceLevel.MID) salary_min, salary_max = _parse_salary(row['salary_range']) description = strip_tags( _clean(row['description']) or _clean(row['quick_description']) or _clean(row['flyer_description']) ) requirements = _clean(row['qualification']) or description[:500] Job.objects.update_or_create( pk=row['id'], defaults={ 'title': title, 'slug': slug, 'company_name': company_name, 'company_slug': slugify(company_name), 'company_initials': ''.join(w[0] for w in company_name.split()[:2]).upper()[:4] or 'CO', 'location': location, 'job_type': job_type, 'experience_level': experience, 'category_id': row['category_id'] if JobCategory.objects.filter(pk=row['category_id']).exists() else None, 'salary_min': salary_min, 'salary_max': salary_max, 'description': description or title, 'responsibilities': description or title, 'requirements': requirements or 'See job description', 'deadline': row['deadline'], 'posted_by_id': posted_by_id, 'is_active': _clean(row['approval_status']) != 'rejected', }, ) count += 1 return count def _import_profiles(self, legacy): cur = legacy.cursor() skill_map = {} cur.execute('SELECT user_id, name FROM profile_store_skill') for row in cur.fetchall(): skill_map.setdefault(row['user_id'], []).append(_clean(row['name'])) exp_map = {} cur.execute('SELECT user_id, job_title, company, work_from, work_to FROM profile_store_workexperience') for row in cur.fetchall(): line = f'{_clean(row["job_title"])} at {_clean(row["company"])} ({_clean(row["work_from"])}-{_clean(row["work_to"])})' exp_map.setdefault(row['user_id'], []).append(line) edu_map = {} cur.execute('SELECT user_id, name, institution, level, year FROM profile_store_academicqualification') for row in cur.fetchall(): line = f'{_clean(row["name"])} - {_clean(row["institution"])} ({row["year"]})' edu_map.setdefault(row['user_id'], []).append(line) cv_map = {} cur.execute('SELECT user_id, file FROM profile_store_cv ORDER BY added_date DESC') for row in cur.fetchall(): uid = row['user_id'] if uid not in cv_map and _clean(row['file']): cv_map[uid] = _clean(row['file']) count = 0 cur.execute("SELECT * FROM accounts_user WHERE account_type='job_seeker' ORDER BY id") for row in cur.fetchall(): uid = row['id'] if not User.objects.filter(pk=uid).exists(): continue full_name = f'{_clean(row["first_name"])} {_clean(row["last_name"])}'.strip() or _clean(row['email']) skills = [] for source in (row['skills'], row['additional_skills'], row['portfolio_skills']): if source: skills.extend(line.strip() for line in str(source).splitlines() if line.strip()) skills.extend(skill_map.get(uid, [])) skills = list(dict.fromkeys(skills)) slug = _unique_slug(SeekerProfile, full_name, pk=uid) cover_url = legacy_media_url(row['cover_image'], settings.MEDIA_URL) avatar_url = legacy_media_url(row['image'], settings.MEDIA_URL) SeekerProfile.objects.update_or_create( user_id=uid, defaults={ 'slug': slug, 'full_name': full_name, 'headline': _clean(row['professional_title']) or _clean(row['current_job_title']) or 'Professional', 'category': _clean(row['industry']) or _clean(row['profession']) or 'General', 'category_slug': slugify(_clean(row['industry']) or 'general'), 'location': _clean(row['city_town']) or _clean(row['state_province']) or 'Zambia', 'skills': '\n'.join(skills), 'bio': _clean(row['short_intro']) or _clean(row['profession']), 'education': '\n'.join(edu_map.get(uid, [])), 'experience_summary': '\n'.join(exp_map.get(uid, [])), 'qualification': _clean(row['highest_qualification']), 'cover_image_url': cover_url, 'avatar_url': avatar_url, 'is_public': True, 'is_featured': bool(row['blue_badge']), 'tier': SeekerProfile.Tier.GOLD if row['blue_badge'] else SeekerProfile.Tier.FREE, }, ) if uid in cv_map: SeekerProfile.objects.filter(user_id=uid).update(cv_file=cv_map[uid]) count += 1 return count def _import_applications(self, legacy): cur = legacy.cursor() file_map = {} cur.execute('SELECT application_id, file FROM store_jobapplicationfile ORDER BY added_date DESC') for row in cur.fetchall(): aid = row['application_id'] if aid not in file_map and _clean(row['file']): file_map[aid] = _clean(row['file']) cur.execute('SELECT * FROM store_jobapplicant ORDER BY id') count = 0 for row in cur.fetchall(): if not Job.objects.filter(pk=row['job_id']).exists(): continue if not User.objects.filter(pk=row['user_id']).exists(): continue status = APPLICATION_STATUS_MAP.get(_clean(row['hired']), JobApplication.Status.PENDING) app, _ = JobApplication.objects.update_or_create( pk=row['id'], defaults={ 'job_id': row['job_id'], 'applicant_id': row['user_id'], 'cover_letter': _clean(row['cover_letter']), 'status': status, }, ) cv_path = file_map.get(row['id']) if cv_path: JobApplication.objects.filter(pk=app.pk).update(cv_file=cv_path) count += 1 return count def _get_plan(self, slug, audience): plan = SubscriptionPlan.objects.filter(slug=slug, audience=audience).first() if plan: return plan fallback = SubscriptionPlan.objects.filter(audience=audience, is_active=True).order_by('sort_order').first() return fallback def _import_subscriptions(self, legacy): seeker_plan_default = self._get_plan( settings.SEEKER_PLUS_PLAN_SLUG, SubscriptionPlan.Audience.SEEKER, ) if not seeker_plan_default: return 0, 0 cur = legacy.cursor() cur.execute( """ SELECT s.user_id, s.status, s.txn, s.expiry_date, s.added_date, u.account_type, p.subscription_type FROM profile_store_subscription s JOIN accounts_user u ON u.id = s.user_id LEFT JOIN profile_store_pricing p ON p.id = s.package_id WHERE s.status IN ('paid', 'delivered') ORDER BY s.expiry_date DESC, s.added_date DESC """ ) best_seeker = {} best_company = {} for row in cur.fetchall(): user_id = row['user_id'] expires_at = _parse_expiry(row['expiry_date']) started_at = _parse_dt(row['added_date']) is_active = expires_at is None or expires_at > timezone.now() payload = { 'expires_at': expires_at, 'started_at': started_at, 'is_active': is_active, 'legacy_txn': _clean(row['txn']), 'subscription_type': _clean(row['subscription_type']) or 'Month', } if row['account_type'] == 'job_seeker': if user_id not in best_seeker: best_seeker[user_id] = payload elif row['account_type'] == 'employer': if user_id not in best_company: best_company[user_id] = payload cur.execute( "SELECT id FROM accounts_user " "WHERE account_type='job_seeker' AND subscription_active=1" ) for row in cur.fetchall(): best_seeker.setdefault(row['id'], { 'expires_at': None, 'started_at': timezone.now(), 'is_active': True, 'legacy_txn': '', 'subscription_type': 'Month', }) seeker_count = 0 for user_id, data in best_seeker.items(): if not User.objects.filter(pk=user_id).exists(): continue SeekerSubscription.objects.update_or_create( user_id=user_id, defaults={ 'plan': seeker_plan_default, 'is_active': data['is_active'], 'expires_at': data['expires_at'], }, ) seeker_count += 1 company_count = 0 default_company_slug = getattr(settings, 'COMPANY_DEFAULT_PLAN_SLUG', 'company-professional') for user_id, data in best_company.items(): company = Company.objects.filter(owner_id=user_id).first() if not company: continue plan_slug = COMPANY_PLAN_BY_PERIOD.get(data['subscription_type'], default_company_slug) plan = self._get_plan(plan_slug, SubscriptionPlan.Audience.COMPANY) if not plan: continue CompanySubscription.objects.update_or_create( company=company, defaults={ 'plan': plan, 'is_active': data['is_active'], 'started_at': data['started_at'], 'expires_at': data['expires_at'], 'legacy_txn': data['legacy_txn'], }, ) tier = COMPANY_TIER_BY_SLUG.get(plan.slug, Company.Tier.PRO) Company.objects.filter(pk=company.pk).update(tier=tier) company_count += 1 return seeker_count, company_count def _import_testimonials(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM testimonial_testimonial ORDER BY id') count = 0 for row in cur.fetchall(): full_name = _clean(row['full_name']) or f'User {row["id"]}' story = _clean(row['description_one']) if not story: continue Testimonial.objects.update_or_create( pk=row['id'], defaults={ 'full_name': full_name, 'role': _clean(row['professional_title']) or 'Member', 'company': _clean(row['company']), 'story': story, 'rating': int(row['rating'] or 5), 'avatar_url': _clean(row['photo']), 'is_approved': bool(row['status']), }, ) count += 1 return count def _import_interests(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM profile_store_interest ORDER BY id') count = 0 for row in cur.fetchall(): if not User.objects.filter(pk=row['user_id']).exists(): continue name = _clean(row['name']) if not name: continue SeekerInterest.objects.update_or_create( pk=row['id'], defaults={ 'user_id': row['user_id'], 'name': name, 'created_at': _parse_dt(row['added_date']), }, ) count += 1 return count def _import_portfolio_stats(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM portfolios_portfoliostat ORDER BY id') count = 0 for row in cur.fetchall(): if not User.objects.filter(pk=row['user_id']).exists(): continue PortfolioStat.objects.update_or_create( user_id=row['user_id'], defaults={ 'total_views': int(row['total_views'] or 0), 'total_likes': int(row['total_likes'] or 0), 'total_comments': int(row['total_comments'] or 0), 'total_items': int(row['total_items'] or 0), 'last_updated': _parse_dt(row['last_updated']) if row['last_updated'] else None, }, ) count += 1 return count def _import_profile_views(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM profile_store_profileview ORDER BY id') count = 0 for row in cur.fetchall(): if not User.objects.filter(pk=row['to_user_id']).exists(): continue viewer_id = row['user_id'] if User.objects.filter(pk=row['user_id']).exists() else None ProfileView.objects.update_or_create( pk=row['id'], defaults={ 'profile_user_id': row['to_user_id'], 'viewer_id': viewer_id, 'account_type': _clean(row['account_type']), 'ip_address': _safe_ip(row['ip']), 'viewed_at': _parse_dt(row['added_date']), }, ) count += 1 return count def _import_profile_messages(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM profile_store_message ORDER BY id') count = 0 for row in cur.fetchall(): if not User.objects.filter(pk=row['to_user_id']).exists(): continue sender_id = row['from_userid'] if User.objects.filter(pk=row['from_userid']).exists() else None ProfileMessage.objects.update_or_create( pk=row['id'], defaults={ 'sender_id': sender_id, 'recipient_id': row['to_user_id'], 'msg_type': _clean(row['msg_type']), 'subject': _clean(row['subject']), 'message': _clean(row['message']), 'view_cv': bool(row['view_cv']), 'is_read': bool(row['read']), 'read_by_sender': bool(row['read_sender']), 'created_at': _parse_dt(row['added_date']), }, ) count += 1 return count def _import_profile_replies(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM profile_store_messagereply ORDER BY id') count = 0 for row in cur.fetchall(): if not ProfileMessage.objects.filter(pk=row['parent_message_id']).exists(): continue if not User.objects.filter(pk=row['to_user_id']).exists(): continue sender_id = row['from_userid'] if User.objects.filter(pk=row['from_userid']).exists() else None ProfileMessageReply.objects.update_or_create( pk=row['id'], defaults={ 'parent_id': row['parent_message_id'], 'sender_id': sender_id, 'recipient_id': row['to_user_id'], 'message': _clean(row['message']), 'image': _clean(row['image']), 'is_read': bool(row['read']), 'created_at': _parse_dt(row['added_date']), }, ) count += 1 return count def _import_notifications(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM profile_store_notification ORDER BY id') count = 0 for row in cur.fetchall(): if not User.objects.filter(pk=row['to_user_id']).exists(): continue sender_id = row['from_userid'] if row['from_userid'] and User.objects.filter(pk=row['from_userid']).exists() else None UserNotification.objects.update_or_create( pk=row['id'], defaults={ 'recipient_id': row['to_user_id'], 'sender_id': sender_id, 'title': _clean(row['title']), 'subject': _clean(row['subject']), 'body': _clean(row['body']), 'message': _clean(row['message']), 'notification_type': _clean(row['notification_type']), 'redirect_url': _clean(row['redirect_url']), 'button_text': _clean(row['button_text']), 'image': _clean(row['image']), 'is_read': bool(row['read']), 'created_at': _parse_dt(row['added_date']), }, ) count += 1 return count def _import_site_contacts(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM chat_contact ORDER BY id') count = 0 for row in cur.fetchall(): user_id = row['user_id_id'] if row['user_id_id'] and User.objects.filter(pk=row['user_id_id']).exists() else None SiteContact.objects.update_or_create( pk=row['id'], defaults={ 'full_name': _clean(row['full_name']) or 'Anonymous', 'email': _clean(row['email']), 'phone': _clean(row['phone_number']), 'country': _clean(row['country']), 'company': _clean(row['company']), 'department': _clean(row['department']), 'item_type': _clean(row['item_type']), 'message': _clean(row['message']), 'link': _clean(row['link']), 'user_id': user_id, 'created_at': _parse_dt(row['added_date']), }, ) count += 1 return count def _import_visitor_traffic(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM accounts_visitortraffic ORDER BY id') count = 0 for row in cur.fetchall(): day = parse_date(str(row['day'])) if not day: continue DailyVisitorCount.objects.update_or_create( pk=row['id'], defaults={ 'day': day, 'count': int(row['count'] or 0), 'updated_at': _parse_dt(row['updated_at']) if row['updated_at'] else None, }, ) count += 1 return count def _import_podcast_categories(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM podcast_podcastcategory ORDER BY id') count = 0 for row in cur.fetchall(): name = _clean(row['name']) or f'Category {row["id"]}' slug = _unique_slug(PodcastCategory, _clean(row['slug']) or name, pk=row['id']) added_by_id = row['added_by_id'] if row['added_by_id'] and User.objects.filter(pk=row['added_by_id']).exists() else None PodcastCategory.objects.update_or_create( pk=row['id'], defaults={ 'name': name, 'slug': slug, 'description': _clean(row['description']), 'thumbnail': _clean(row['thumbnail']), 'added_by_id': added_by_id, 'created_at': _parse_dt(row['added_date']), }, ) count += 1 return count def _import_job_views(self, legacy): valid_jobs = set(Job.objects.values_list('pk', flat=True)) valid_users = set(User.objects.values_list('pk', flat=True)) JobView.objects.all().delete() count = 0 batch = [] batch_size = 2000 def flush(): nonlocal count, batch if batch: JobView.objects.bulk_create(batch, batch_size=batch_size) count += len(batch) batch = [] cur = legacy.cursor() for table, is_unique in ( ('store_itemviewcountall', False), ('store_itemviewcountunique', True), ): cur.execute(f'SELECT * FROM {table} ORDER BY id') for row in cur.fetchall(): if row['item_id'] not in valid_jobs: continue user_id = row['user_id'] if row['user_id'] in valid_users else None batch.append(JobView( job_id=row['item_id'], user_id=user_id, ip_address=_safe_ip(row['ip_address']), is_unique=is_unique, viewed_at=_parse_dt(row['added_date']), )) if len(batch) >= batch_size: flush() flush() view_totals = JobView.objects.filter(is_unique=False).values('job_id').annotate(total=Count('id')) for entry in view_totals: Job.objects.filter(pk=entry['job_id']).update(views_count=entry['total']) return count def _import_payment_methods(self, legacy): count = 0 for item in DEFAULT_PAYMENT_METHODS: PaymentMethod.objects.update_or_create( slug=item['slug'], defaults={ 'name': item['name'], 'description': item['description'], 'sort_order': item['sort_order'], 'is_active': True, }, ) count += 1 cur = legacy.cursor() cur.execute('SELECT * FROM business_paymentmethod ORDER BY id') for row in cur.fetchall(): name = _clean(row['name']) if not name: continue slug = _unique_slug(PaymentMethod, name) PaymentMethod.objects.update_or_create( legacy_id=row['id'], defaults={ 'slug': slug, 'name': name, 'is_active': True, 'sort_order': 10 + row['id'], }, ) count += 1 return count def _legacy_pricing_plan_map(self, legacy): cur = legacy.cursor() cur.execute('SELECT * FROM profile_store_pricing ORDER BY id') mapping = {} seeker_default = self._get_plan( settings.SEEKER_PLUS_PLAN_SLUG, SubscriptionPlan.Audience.SEEKER, ) company_default = self._get_plan( getattr(settings, 'COMPANY_DEFAULT_PLAN_SLUG', 'company-professional'), SubscriptionPlan.Audience.COMPANY, ) for row in cur.fetchall(): user_type = _clean(row['user_type']) or 'job_seeker' if user_type == 'job_seeker': plan = seeker_default else: plan_slug = COMPANY_PLAN_BY_PERIOD.get( _clean(row['subscription_type']), getattr(settings, 'COMPANY_DEFAULT_PLAN_SLUG', 'company-professional'), ) plan = self._get_plan(plan_slug, SubscriptionPlan.Audience.COMPANY) or company_default if plan: mapping[row['id']] = (plan, int(row['price'] or 0)) return mapping def _import_subscription_payments(self, legacy): pricing_map = self._legacy_pricing_plan_map(legacy) payment_methods = { method.slug: method for method in PaymentMethod.objects.all() } cur = legacy.cursor() cur.execute( """ SELECT s.id, s.user_id, s.txn, s.status, s.provider_txn, s.expiry_date, s.added_date, s.package_id, u.account_type FROM profile_store_subscription s JOIN accounts_user u ON u.id = s.user_id ORDER BY s.id """ ) count = 0 for row in cur.fetchall(): if not User.objects.filter(pk=row['user_id']).exists(): continue plan, amount = pricing_map.get(row['package_id'], (None, 0)) if not plan: audience = ( SubscriptionPlan.Audience.SEEKER if row['account_type'] == 'job_seeker' else SubscriptionPlan.Audience.COMPANY ) plan = self._get_plan( settings.SEEKER_PLUS_PLAN_SLUG if audience == SubscriptionPlan.Audience.SEEKER else getattr(settings, 'COMPANY_DEFAULT_PLAN_SLUG', 'company-professional'), audience, ) if not plan: continue method_slug = _infer_payment_method_slug(row['txn'], row['provider_txn']) payment_method = payment_methods.get(method_slug) status = LEGACY_PAYMENT_STATUS.get( _clean(row['status']), SubscriptionPayment.Status.PENDING, ) expires_at = _parse_expiry(row['expiry_date']) paid_at = _parse_dt(row['added_date']) if status in ( SubscriptionPayment.Status.PAID, SubscriptionPayment.Status.DELIVERED, ) else None company = ( Company.objects.filter(owner_id=row['user_id']).first() if row['account_type'] == 'employer' else None ) SubscriptionPayment.objects.update_or_create( legacy_subscription_id=row['id'], defaults={ 'user_id': row['user_id'], 'company': company, 'payment_method': payment_method, 'plan': plan, 'amount': amount, 'txn_reference': _clean(row['txn']), 'provider_reference': _clean(row['provider_txn']), 'status': status, 'paid_at': paid_at, 'expires_at': expires_at, }, ) count += 1 return count def _sync_subscription_benefits(self): for subscription in SeekerSubscription.objects.select_related('user', 'plan'): sync_seeker_subscription_benefits(subscription) for subscription in CompanySubscription.objects.select_related('company', 'plan'): sync_company_subscription_benefits(subscription) def _copy_media(self, media_source): import shutil source = Path(media_source) if not source.exists(): self.stdout.write(self.style.WARNING(f'Media source not found: {source}')) return 0 target = Path(settings.MEDIA_ROOT) target.mkdir(parents=True, exist_ok=True) copied = 0 for src_file in source.rglob('*'): if not src_file.is_file(): continue rel = src_file.relative_to(source) dest = target / rel dest.parent.mkdir(parents=True, exist_ok=True) if not dest.exists() or src_file.stat().st_mtime > dest.stat().st_mtime: shutil.copy2(src_file, dest) copied += 1 return copied