#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Location-Centric JobPosting Optimizer (CTR DOMINATOR EDITION)
=============================================================
This version discards "Standard" SEO in favor of "High CTR" tactics.
It uses visual interrupts, benefit-driven copy, and psychological pricing.
KEY UPGRADES:
1. VISUAL INTERRUPTS: Uses [], (), and | to break scanning patterns.
2. BENEFIT-FIRST TITLES: Puts "Urgent", "Flexible", "Weekly Pay" first.
3. INTELLIGENT CONTEXT: Detects roles (Driver, Nurse, Retail) and changes the sales pitch.
4. PRETTY MATH: Rounds hourly rates to quarters ($0.00, $0.25, $0.50, $0.75).
Usage:
python3 10-11-short-title-v2.py -i input.ndjson -o output.ndjson
"""
import argparse
import json
import random
import re
import logging
import locale
import hashlib
from tqdm import tqdm
from bs4 import BeautifulSoup
from datetime import datetime, timedelta, timezone
# ==========================
# Locale initialization
# ==========================
try:
locale.setlocale(locale.LC_ALL, "")
except locale.Error:
try:
locale.setlocale(locale.LC_ALL, "en_US.UTF-8")
except locale.Error:
locale.setlocale(locale.LC_ALL, "C.UTF-8")
# ==========================
# Constants & Defaults
# ==========================
MAX_TITLE_LEN_BASE = 68
DESIRED_DEFAULT_FALLBACK_LOGO_URL = "https://walmart.uscareers.co.com/logo.png"
DEFAULT_CURRENCY = "USD"
CONTACT_CTA = ''
DEFAULT_FULL_TIME_HOURS_PER_WEEK = 40
DEFAULT_PART_TIME_HOURS_PER_WEEK = 20 # Reduced for better hourly perception
WEEKS_PER_YEAR = 52
MONTHS_PER_YEAR = 12
SALARY_ADJUSTMENT_FACTORS = [1.0, 0.95, 1.05] # Tightened range
WEIGHTED_SALARY_ADJUSTMENT_FACTORS = [1.0, 1.0] + SALARY_ADJUSTMENT_FACTORS
SALARY_UNIT_MAP = {
'HOUR': 'hr', 'HOURLY': 'hr', 'DAY': 'day', 'DAILY': 'day',
'WEEK': 'wk', 'WEEKLY': 'wk', 'MONTH': 'mo', 'MONTHLY': 'mo',
'YEAR': 'yr', 'ANNUAL': 'yr', 'ANNUALLY': 'yr', 'PROJECT': 'project'
}
NORMALIZED_SALARY_UNITS = {
"HOUR": "HOUR", "HOURLY": "HOUR", "DAY": "PROJECT", "DAILY": "PROJECT",
"WEEK": "WEEK", "WEEKLY": "WEEK", "MONTH": "MONTH", "MONTHLY": "MONTH",
"YEAR": "YEAR", "ANNUAL": "YEAR", "ANNUALLY": "YEAR", "PROJECT": "PROJECT"
}
EXPERIENCE_LEVEL_MONTHS = {
"Entry-Level": (0, 12),
"Mid-Level": (13, 60),
"Senior-Level": (61, float('inf'))
}
EXPERIENCE_LEVEL_KEYWORDS = {
"Entry-Level": ["entry level", "no experience", "graduate", "junior", "trainee", "intern"],
"Mid-Level": ["mid level", "intermediate", "associate", "experienced"],
"Senior-Level": ["senior", "lead", "principal", "expert", "staff level", "manager"]
}
RECENT_POST_DAYS = 2
CLOSING_SOON_DAYS = 7
# ==========================
# INTELLIGENCE MAPS (New)
# ==========================
ROLE_CONTEXT_MAP = {
"warehouse": {
"hook": "Immediate Start & Weekly Pay",
"synonym": "Warehouse Associate",
"vibe": "active"
},
"driver": {
"hook": "Earn Cash on Your Schedule",
"synonym": "Driver Opportunity",
"vibe": "gig"
},
"delivery": {
"hook": "Flexible Delivery Routes",
"synonym": "Courier",
"vibe": "gig"
},
"nurse": {
"hook": "Flexible Shifts (Day/Night)",
"synonym": "Nursing Role",
"vibe": "care"
},
"retail": {
"hook": "Weekend & Evening Shifts Available",
"synonym": "Retail Associate",
"vibe": "customer"
},
"sales": {
"hook": "Uncapped Commission Potential",
"synonym": "Sales Representative",
"vibe": "money"
},
"server": {
"hook": "Tips + Hourly Pay",
"synonym": "Hospitality Role",
"vibe": "customer"
},
"customer service": {
"hook": "Remote/Hybrid Options May Apply",
"synonym": "Support Specialist",
"vibe": "office"
}
}
# ==================================================
# HIGH-CTR TEMPLATES (Completely Redesigned)
# ==================================================
GEO_HEADER_TEMPLATES = [
"Hiring in {city}: {role}",
"{city} Opportunity: {role}",
"Job Overview: {role} ({city})",
"Role Based in {city}",
"Now Hiring: {role}"
]
SUB_HEADER_TEMPLATES = {
"primary": ["Why This Role?", "Position Snapshot", "Quick Facts", "The Opportunity"],
"secondary": ["Pay & Perks", "What We Offer", "Compensation"],
"tertiary": ["Your Duties", "What You Will Do", "The Daily Grind"]
}
# ==================================================
# PERSUASIVE DESCRIPTION BLOCKS
# ==================================================
GEO_TONE_TEMPLATES = {
"FULL_TIME": {
'informational': {
'intro': [
"Full-Time Career in {city}: {role_for_summary}",
"Position: {role_for_summary} (Full-Time)"
],
'company': ["Employer: {company_for_summary}"],
'location': ["Based in: {city}, {state}"],
'salary': ["Compensation: {salary_primary_display}{converted_salary_suffix}"],
'experience_entry': ["Experience: Entry-Level / Training Provided."],
'experience_mid': ["Experience: Associate Level."],
'experience_senior': ["Experience: Senior/Lead Level."],
'urgency_new': ["Status: Just Posted."],
'urgency_closing': ["Status: Final Call."],
'focus': ["Join our {city} team as a {role_for_summary}."],
'benefits': ["Full benefits package included.", "Standard full-time benefits."],
'value_props': ["Stable career path.", "Long-term opportunity."]
}
},
# --- AGGRESSIVE PART-TIME (GIG/LIFESTYLE TONE) ---
"PART_TIME": {
'informational': {
'intro': [
"OPENING in {city}: Part-Time {role_for_summary}",
"FLEXIBLE ROLE: {role_for_summary} ({city})",
"URGENT HIRE: {role_for_summary} (Part-Time)",
"OPPORTUNITY: Earn extra income as a {role_for_summary}"
],
'company': [
"Hiring: {company_for_summary}",
"Team: {company_for_summary} ({city} Branch)"
],
'location': [
"Location: {city}, {state} (Local Candidates Only)",
"Worksite: {city} Area"
],
'salary': [
"Pay Rate: {salary_primary_display}{converted_salary_suffix}",
"Est. Earnings: {salary_primary_display}{converted_salary_suffix}",
"Starting Pay: {salary_primary_display}{converted_salary_suffix}"
],
'experience_entry': [
"Experience: No Experience Needed - Will Train!",
"Perfect For: Students, Grads, or First-Time Job Seekers.",
"Start: Immediate start available."
],
'experience_mid': ["Experience: Some prior experience preferred."],
'experience_senior': ["Experience: Experienced candidates needed."],
'urgency_new': [
"Status: ⚡ New Listing.",
"Timeline: Interviews this week in {city}."
],
'urgency_closing': ["Status: ⏳ Closing Soon."],
'focus': [
"We need a reliable {role_for_summary} in {city}.",
"Looking for flexibility? Join us as a {role_for_summary}."
],
'benefits': [
"Flexible scheduling.",
"Employee discounts.",
"Fast-paced environment."
],
'value_props': [
"Great for supplemental income.",
"Work around your school/life schedule.",
"Weekly pay options often available.",
"Join a top-rated team in {city}."
]
}
},
"FLEXIBLE": {
'informational': {
'intro': [
"GIG / FLEX ROLE: {role_for_summary} ({city})",
"Make Your Schedule: {role_for_summary}"
],
'company': ["Platform/Employer: {company_for_summary}"],
'location': ["Territory: {city}, {state}"],
'salary': ["Potential Earnings: {salary_primary_display}"],
'experience_entry': ["Reqs: minimal requirements."],
'experience_mid': ["Reqs: Prior experience needed."],
'experience_senior': ["Reqs: Professional experience needed."],
'urgency_new': ["Status: Accepting new drivers/workers."],
'urgency_closing': ["Status: Limited spots."],
'focus': ["Earn on your own terms in {city}."],
'benefits': ["Be your own boss.", "Instant pay options."],
'value_props': ["The ultimate side hustle.", "Cash gig opportunity."]
}
}
}
HIRING_SYNS = ["Hiring", "Start ASAP", "Now Hiring", "Apply Now", "Urgent", "Open Now"]
ENTRY_SYNS = ["Entry Level", "No Experience", "Junior", "Training", "Student"]
URGENCY_TAGS_NEW_TITLE = ["New", "Just Posted", "Fresh"]
URGENCY_TAGS_CLOSING_TITLE = ["Closing", "Urgent", "Last Chance"]
# ==================================================
# VISUAL INTERRUPT TITLE TEMPLATES
# ==================================================
GEO_FOCUSED_TITLE_TEMPLATES = {
"FULL_TIME": [
"{city} Full-Time: {role}",
"{role} (FT) - {city}",
"[Full-Time] {role} - {city}, {state}",
"{city} Career: {role} (Full-Time)",
"Hiring: {role} (Full-Time) in {city}",
"{role} - {company} ({city})",
"{city} Opening: {role} (FT)"
],
# --- AGGRESSIVE PART-TIME VARIATIONS ---
"PART_TIME": [
# BRACKET METHOD (High Visibility)
"[Part-Time] {role} - {city}",
"({city}) Part-Time: {role}",
"[{urgency_tag}] {role} - {city} (PT)",
# BENEFIT FIRST
"Start ASAP: {role} ({city})",
"Flexible Hours: {role} - {city}",
"Student Friendly: {role} ({city})",
"Weekend Shift: {role} - {city}",
"Evening Role: {role} ({city})",
# INCOME FIRST (If salary avail)
"{salary_compact} {role} - {city} (PT)",
"Earn Extra Cash: {role} ({city})",
# LOCAL URGENCY
"{city} Needs: {role} (Part-Time)",
"Urgent Hire in {city}: {role}",
"{city} Job: {role} (No Exp Req*)",
# DIRECT
"{role} needed in {city} (Apply Now)",
"Hiring in {city}: {role} (Part-Time)",
"{role} (Part-Time) @ {company} - {city}"
],
"FLEXIBLE": [
"[Flex] {role} - {city}",
"{role} (Gig) - Earn in {city}",
"Make Your Schedule: {role} ({city})",
"{city} Side Hustle: {role}",
"({city}) {role} - Contract/Gig",
"Driver/Gig: {role} in {city}",
"Start Today: {role} ({city})"
]
}
# ==========================
# Argparse
# ==========================
def parse_args():
p = argparse.ArgumentParser(description="Optimize Local JobPosting NDJSON with a strong Part-Time & Location focus.")
p.add_argument('-i','--input', default='all-schemas.ndjson', help='Input NDJSON file')
p.add_argument('-o','--output', default='schema.ndjson', help='Output NDJSON file')
p.add_argument('--seed', type=int, default=None, help='Seed for reproducible randomness')
p.add_argument('--logo_cdn', default=DESIRED_DEFAULT_FALLBACK_LOGO_URL, help='Default fallback logo URL')
p.add_argument('--currency', default=DEFAULT_CURRENCY, help='Default currency')
p.add_argument('-v','--verbose', action='store_true', help='Enable debug logging')
p.add_argument('--full_time_hours', type=int, default=DEFAULT_FULL_TIME_HOURS_PER_WEEK, help="Standard hours for full-time salary conversions.")
p.add_argument('--part_time_hours', type=int, default=DEFAULT_PART_TIME_HOURS_PER_WEEK, help="Standard hours for part-time salary conversions.")
p.add_argument('--enable_salary_adjustment', action='store_true', help='Enable dynamic salary value adjustments.')
title_group = p.add_mutually_exclusive_group()
title_group.add_argument('--no-change-in-title', action='store_true', help='Do not improve the job title.')
title_group.add_argument('--slightly-improve-title', action='store_true', help='Apply only minor, geo-focused improvements.')
return p.parse_args()
# ==========================
# Small utilities
# ==========================
def title_case(s: str) -> str:
if not s:
return ""
words = [w.capitalize() if not (w.isupper() and len(w) > 1) else w for w in s.split()]
return " ".join(words)
def to_dhaka_offset(dt: str) -> str:
if not dt: return ""
try:
parsed_dt = datetime.fromisoformat(dt.replace('Z', '+00:00'))
dhaka_tz = timezone(timedelta(hours=6))
return parsed_dt.astimezone(dhaka_tz).isoformat()
except ValueError: return dt
def to_midnight(dt_str: str) -> str:
if not dt_str: return ""
try:
dt_obj = datetime.fromisoformat(dt_str.replace('Z', '+00:00'))
target_tz = dt_obj.tzinfo or timezone(timedelta(hours=6))
dt_obj_midnight = dt_obj.astimezone(target_tz).replace(hour=23, minute=59, second=59, microsecond=0)
return dt_obj_midnight.isoformat()
except (IndexError, ValueError): return dt_str
def normalize_url(url: str) -> str:
if not url: return ""
url = url.split('?',1)[0].split('#',1)[0]
if url.startswith('//'): url = 'https://' + url[2:]
elif url.startswith('http://'): url = 'https://' + url[7:]
elif not url.startswith('https://'): url = 'https://' + url.lstrip('/')
if not re.search(r'\.\w{2,5}(?:/)?$', url.split('/')[-1]) and not url.endswith('/'):
url += '/'
return url
def get_currency_symbol(currency_code: str) -> str:
symbols = {"USD":"$", "EUR":"€", "GBP":"£", "JPY":"¥", "CAD":"CA$", "AUD":"A$", "INR":"₹", "BDT":"৳"}
return symbols.get(str(currency_code).upper(), "$")
# ==========================
# Intelligence helpers
# ==========================
def get_primary_skill(skills_value) -> str:
if not skills_value: return ""
processed = ""
if isinstance(skills_value, str):
processed = skills_value
elif isinstance(skills_value, list):
string_skills = []
for item in skills_value:
if isinstance(item, str) and item.strip():
string_skills.append(item.strip())
processed = ", ".join(string_skills)
else:
processed = str(skills_value).strip()
if not processed: return ""
try:
return next((s.strip() for s in re.split(r'[,;/]', processed) if s.strip()), "")
except TypeError: return ""
def get_location_details(rec: dict) -> tuple[str, str, str]:
job_loc_data = rec.get('jobLocation')
city, state, country = "", "", "US"
if isinstance(job_loc_data, dict):
address_data = job_loc_data.get('address')
if isinstance(address_data, dict):
city = address_data.get('addressLocality', '')
state = address_data.get('addressRegion', '')
country = address_data.get('addressCountry', country)
elif isinstance(job_loc_data, list) and job_loc_data:
first_loc = job_loc_data[0]
if isinstance(first_loc, dict):
address_data = first_loc.get('address')
if isinstance(address_data, dict):
city = address_data.get('addressLocality', '')
state = address_data.get('addressRegion', '')
country = address_data.get('addressCountry', country)
if not city and state: city = state
return str(city), str(state), str(country)
def get_employment_types_info(rec: dict) -> dict:
"""Aggressive Classification: Guilty of Part-Time until proven Full-Time."""
et_input = rec.get('employmentType')
normalized_types = []
if isinstance(et_input, list):
for item in et_input:
if isinstance(item, str) and item.strip():
norm_item = item.replace('_', '-').strip().lower()
normalized_types.append(title_case(norm_item.replace(" time", "-Time")))
elif isinstance(et_input, str) and et_input.strip():
norm_item = et_input.replace('_', '-').strip().lower()
normalized_types.append(title_case(norm_item.replace(" time", "-Time")))
unique_types = sorted(list(set(normalized_types)))
is_full_time = "Full-Time" in unique_types
is_part_time = "Part-Time" in unique_types
template_key = "PART_TIME"
chosen_for_title = "Part-Time"
chosen_for_description = "Part-Time" # Default for description logic
if is_full_time and not is_part_time:
template_key = "FULL_TIME"
chosen_for_title = "Full-Time"
chosen_for_description = "Full-Time"
elif is_part_time:
template_key = "PART_TIME"
chosen_for_title = "Part-Time"
chosen_for_description = "Part-Time"
elif any(x in unique_types for x in ["Contract", "Temporary", "Seasonal", "Gig"]):
template_key = "FLEXIBLE"
chosen_for_title = "Flexible"
chosen_for_description = "Flexible"
else:
# Default Assumption for missing data
template_key = "PART_TIME"
chosen_for_title = "Part-Time"
chosen_for_description = "Part-Time"
schema_list = [t.upper().replace('-', '_') for t in unique_types] if unique_types else ["PART_TIME"]
return {
'title_display': chosen_for_title,
'chosen_for_description': chosen_for_description, # Restored missing key
'schema_list': schema_list,
'template_key': template_key
}
# --- FIX: RESTORED schema_list key ---
def get_industries_info(rec: dict) -> dict:
industry_input = rec.get('industry')
processed_industries = []
if isinstance(industry_input, list):
for item in industry_input:
if isinstance(item, str) and item.strip():
processed_industries.append(title_case(item.replace('&', 'and').strip()))
elif isinstance(industry_input, str) and industry_input.strip():
processed_industries.append(title_case(industry_input.replace('&', 'and').strip()))
unique_industries = sorted(list(set(processed_industries)))
if not unique_industries:
# FIXED: Added missing 'schema_list' key here
return {'display_list': [], 'title_display': "", 'schema_list': []}
return {
'display_list': unique_industries,
'title_display': random.choice(unique_industries),
# FIXED: Added missing 'schema_list' key here
'schema_list': unique_industries
}
def get_experience_level_info(rec: dict) -> dict:
exp_req = rec.get("experienceRequirements", {})
months_exp = None
level_tag = ""
description_tag = ""
if isinstance(exp_req, dict):
months_str = exp_req.get("monthsOfExperience")
desc_str = exp_req.get("description", "").lower()
try: months_exp = int(months_str) if months_str is not None else None
except: pass
if months_exp is not None:
for level, (min_m, max_m) in EXPERIENCE_LEVEL_MONTHS.items():
if min_m <= months_exp <= max_m:
description_tag = level
level_tag = level.split('-')[0]
break
else:
for level, keywords in EXPERIENCE_LEVEL_KEYWORDS.items():
if any(kw in desc_str for kw in keywords):
description_tag = level
level_tag = level.split('-')[0]
break
if not level_tag and any(syn.lower() in desc_str for syn in ENTRY_SYNS):
description_tag, level_tag = "Entry-Level", "Entry"
return {"title_tag": level_tag, "description_tag": description_tag}
def get_job_urgency_tags(date_posted_str: str, valid_through_str: str, rec_id: str) -> dict:
urgency = {"title_tag": "", "description_key": None}
now_utc = datetime.now(timezone.utc)
if date_posted_str:
try:
posted_dt = datetime.fromisoformat(date_posted_str.replace('Z', '+00:00')).astimezone(timezone.utc)
if (now_utc - posted_dt).days <= RECENT_POST_DAYS:
urgency.update({"title_tag": random.choice(URGENCY_TAGS_NEW_TITLE), "description_key": "new"})
except ValueError: pass
if valid_through_str:
try:
valid_dt = datetime.fromisoformat(valid_through_str.replace('Z', '+00:00')).astimezone(timezone.utc)
if timedelta(days=0) <= (valid_dt - now_utc) <= timedelta(days=CLOSING_SOON_DAYS):
urgency.update({"title_tag": random.choice(URGENCY_TAGS_CLOSING_TITLE), "description_key": "closing"})
except ValueError: pass
return urgency
def to_k_notation(num_val: float, currency_symbol: str) -> str:
if abs(num_val) >= 1000:
k_val = num_val / 1000.0
return f"{currency_symbol}{k_val:.1f}k".replace(".0k", "k")
return f"{currency_symbol}{int(num_val)}"
def format_salary_details(rec: dict, currency_symbol: str = "$", enable_dynamic_adjustment: bool = False,
chosen_emp_type: str = "Full-Time", full_time_hours: int = DEFAULT_FULL_TIME_HOURS_PER_WEEK,
part_time_hours: int = DEFAULT_PART_TIME_HOURS_PER_WEEK) -> dict:
base = rec.get('baseSalary', {})
val_obj = base.get('value', {})
if not isinstance(base, dict): base = {}
if not isinstance(val_obj, dict): val_obj = {}
minv_raw = val_obj.get('minValue', base.get('minValue'))
maxv_raw = val_obj.get('maxValue', base.get('maxValue'))
unit_raw = str(val_obj.get('unitText', base.get('unitText', ''))).upper()
primary_unit_normalized = NORMALIZED_SALARY_UNITS.get(unit_raw, "PROJECT")
def parse_salary_value(s_val):
if s_val is None: return None
if isinstance(s_val, (int, float)): return float(s_val)
if isinstance(s_val, str):
s_val_cleaned = str(s_val).replace(currency_symbol, '').replace(',', '').strip()
if "negotiable" in s_val_cleaned.lower() or not s_val_cleaned: return "Negotiable"
try: return float(s_val_cleaned)
except ValueError: return None
return None
# --- INTELLIGENT UPGRADE: PRETTY ROUNDING ---
def pretty_round(val):
"""Rounds to nearest 0.25 for psychological pricing."""
if val is None: return None
return round(val * 4) / 4
min_val_num = parse_salary_value(minv_raw)
max_val_num = parse_salary_value(maxv_raw)
if min_val_num == "Negotiable" or max_val_num == "Negotiable":
return {"primary_display": "Negotiable", "is_negotiable": True, "conversions": {}}
if min_val_num is None and max_val_num is None:
return {"primary_display": "", "is_negotiable": False, "conversions": {}}
if enable_dynamic_adjustment:
adjustment_factor = random.choice(WEIGHTED_SALARY_ADJUSTMENT_FACTORS)
if isinstance(min_val_num, (int, float)): min_val_num *= adjustment_factor
if isinstance(max_val_num, (int, float)): max_val_num *= adjustment_factor
primary_value_for_conversion = None
primary_display_val = "Error"
if min_val_num is not None and max_val_num is not None:
primary_value_for_conversion = (min_val_num + max_val_num) / 2.0
primary_display_val = f"{to_k_notation(min_val_num, currency_symbol)}-{to_k_notation(max_val_num, currency_symbol)}"
elif max_val_num is not None:
primary_value_for_conversion = max_val_num
primary_display_val = f"Up to {to_k_notation(max_val_num, currency_symbol)}"
elif min_val_num is not None:
primary_value_for_conversion = min_val_num
primary_display_val = to_k_notation(min_val_num, currency_symbol)
primary_unit_display = SALARY_UNIT_MAP.get(primary_unit_normalized, "")
primary_salary_str = f"{primary_display_val}{'/' + primary_unit_display if primary_unit_display and primary_unit_display != 'project' else ''}"
conversions = {}
hours_per_week = part_time_hours if "part-time" in chosen_emp_type.lower() else full_time_hours
if primary_value_for_conversion is not None and primary_unit_normalized != "PROJECT":
annual_equiv = None
if primary_unit_normalized == "HOUR":
annual_equiv = primary_value_for_conversion * hours_per_week * WEEKS_PER_YEAR
elif primary_unit_normalized == "WEEK":
annual_equiv = primary_value_for_conversion * WEEKS_PER_YEAR
elif primary_unit_normalized == "MONTH":
annual_equiv = primary_value_for_conversion * MONTHS_PER_YEAR
elif primary_unit_normalized == "YEAR":
annual_equiv = primary_value_for_conversion
if annual_equiv is not None:
# Calculate Hourly separately for prettification
if hours_per_week > 0:
hourly_val = (annual_equiv / WEEKS_PER_YEAR) / hours_per_week
pretty_hourly = pretty_round(hourly_val)
conversions["HOUR"] = f"{currency_symbol}{pretty_hourly:.2f}/hr"
# Other units
if primary_unit_normalized != "YEAR":
conversions["YEAR"] = f"{to_k_notation(annual_equiv, currency_symbol)}/yr"
# Add WEEK if needed
conversions["WEEK"] = f"{to_k_notation(annual_equiv / WEEKS_PER_YEAR, currency_symbol)}/wk"
return {
"primary_display": primary_salary_str,
"primary_raw_min": min_val_num,
"primary_raw_max": max_val_num,
"primary_unit_normalized": primary_unit_normalized,
"is_negotiable": False,
"conversions": conversions
}
# ==========================
# Content assembly
# ==========================
def clean_role_and_company(original_title: str, org_name_from_ho: str) -> tuple[str, str]:
org_name = str(org_name_from_ho or "").strip()
role = re.sub(r'\s*\(.*?[mfvdix].*?\)\s*', '', str(original_title), flags=re.IGNORECASE).strip()
role = re.sub(r"\s+jobs?\b", "", role, flags=re.IGNORECASE).strip()
# Clean common fluff to get the RAW role for template injection
fluff = [r'\bpart[- ]?time\b', r'\bfull[- ]?time\b', r'\bhiring\b', r'\bneeded\b', r'\burgent\b', r'\bimmediate\b']
for f in fluff:
role = re.sub(f, '', role, flags=re.IGNORECASE)
company, final_role = org_name, role
if not company:
preps = ["at", "for", "with"]
for prep in preps:
match = re.search(rf"^(.*?)\s+{re.escape(prep)}\s+([\w\s.,'&()-]+)$", role, flags=re.IGNORECASE)
if match and 2 <= len(match.group(2).split()) <= 5:
company, final_role = match.group(2).strip(), match.group(1).strip()
break
if company:
final_role = re.sub(rf"\s*\b{re.escape(company)}\b", "", final_role, flags=re.IGNORECASE).strip(" -|,")
if not company: company = "Local Employer"
if not final_role: final_role = "Team Member"
return final_role.strip(), company.strip()
def geo_context_enrichment(html_block: str, role: str, company: str, primary_skill: str, city: str, state: str, industry_display: str) -> str:
soup = BeautifulSoup(html_block or "", 'html.parser')
text = soup.decode_contents() if html_block else ""
# Use the Intelligence Map if available
role_lower = role.lower()
smart_vibe = "standard"
for key, ctx in ROLE_CONTEXT_MAP.items():
if key in role_lower:
smart_vibe = ctx['vibe']
break
synonyms = {r"\bcompany\b": "team", r"\bemployees\b": "staff"}
for pat, repl in synonyms.items():
text = re.sub(pat, repl, text, flags=re.IGNORECASE)
# Vibe-based Stitches
stitches = []
if smart_vibe == "gig" or smart_vibe == "active":
stitches.append(f"{city} needs reliable people like you.")
stitches.append(f"Start earning quickly with {company}.")
else:
stitches.append(f"Work in {city}, {state}.")
stitches.append(f"Join {company} and grow your skills.")
ul = soup.find('ul')
if ul:
for item in stitches: ul.append(BeautifulSoup(item, 'html.parser'))
else:
ul_new = BeautifulSoup("", 'html.parser')
for item in stitches: ul_new.ul.append(BeautifulSoup(item, 'html.parser'))
soup.append(ul_new)
return soup.decode_contents()
def create_geo_targeted_summary(rec: dict, primary_skill: str, salary_details: dict, job_urgency: dict, exp_level_info: dict, industries_info: dict, emp_types_info: dict) -> str:
role_for_summary, company_for_summary = clean_role_and_company(rec.get('title',''), rec.get('hiringOrganization',{}).get('name',''))
city, state, _ = get_location_details(rec)
# --- INTELLIGENT CONTEXT LOOKUP ---
role_lower = role_for_summary.lower()
smart_context = None
for key, ctx in ROLE_CONTEXT_MAP.items():
if key in role_lower:
smart_context = ctx
break
if smart_context: role_for_summary = smart_context['synonym']
emp_type_key = emp_types_info.get('template_key', 'PART_TIME')
tone_bank = GEO_TONE_TEMPLATES.get(emp_type_key, GEO_TONE_TEMPLATES['PART_TIME'])
active = tone_bank['informational']
salary_primary_display = salary_details.get("primary_display", "Negotiable") if salary_details else "Negotiable"
converted_salary_suffix = ""
if salary_details and salary_details.get("conversions"):
if emp_type_key == "PART_TIME" and "HOUR" in salary_details["conversions"]:
best_conv = salary_details["conversions"]["HOUR"]
else:
best_conv = salary_details["conversions"].get("YEAR") or random.choice(list(salary_details["conversions"].values()))
converted_salary_suffix = f" ({best_conv})"
data = {
"role_for_summary": title_case(role_for_summary),
"company_for_summary": company_for_summary,
"city": city,
"state": state,
"salary_primary_display": salary_primary_display,
"converted_salary_suffix": converted_salary_suffix,
"primary_skill": primary_skill or "your skills"
}
def phrase(key): return random.choice(active[key]).format(**data)
job_details = [phrase('intro'), phrase('company'), phrase('location')]
urgency_key = job_urgency.get("description_key")
if urgency_key: job_details.append(phrase(f'urgency_{urgency_key}'))
exp_level = exp_level_info.get("description_tag")
if exp_level == "Entry-Level": job_details.append(phrase('experience_entry'))
elif exp_level == "Mid-Level": job_details.append(phrase('experience_mid'))
elif exp_level == "Senior-Level": job_details.append(phrase('experience_senior'))
comp_benefits = [phrase('salary'), f"Perks: {random.choice(active['benefits'])}"]
if smart_context:
value_props = [f"Highlight: {smart_context['hook']} in {city}."]
else:
value_props = [f"{random.choice(active['value_props']).format(city=city)}"]
role_focus = [random.choice(active['focus']).format(**data)]
summary_html = f"{random.choice(GEO_HEADER_TEMPLATES).format(city=city, state=state, role=title_case(role_for_summary))}
"
# Simple, clean structure for mobile readability
summary_html += (
f"{random.choice(SUB_HEADER_TEMPLATES['primary'])}
" +
"".join(f"- {p}
" for p in job_details) +
"
"
)
summary_html += (
f"{random.choice(SUB_HEADER_TEMPLATES['secondary'])}
" +
"".join(f"- {p}
" for p in (comp_benefits + value_props)) +
"
"
)
summary_html = geo_context_enrichment(
summary_html, title_case(role_for_summary), company_for_summary,
primary_skill, city, state, ""
)
return BeautifulSoup(summary_html, 'html.parser').decode_contents().strip()
# ==========================
# Title building
# ==========================
def enforce_length(title: str, max_len: int) -> str:
if len(title) <= max_len: return title.strip()
shortened = re.sub(r'\s*\([^)]*\)\s*$', '', title).strip()
if len(shortened) <= max_len: return shortened
return title[:max_len-3].strip() + "..."
def generate_location_focused_title(rec: dict, primary_skill: str, salary_details: dict, job_urgency:dict,
exp_level_info:dict, emp_types_info:dict, industries_info:dict,
dynamic_max_len:int) -> str:
ho_name = rec.get('hiringOrganization', {}).get('name', '')
cleaned_role, company_name = clean_role_and_company(rec.get('title', rec.get('name','')), ho_name)
city, state, _ = get_location_details(rec)
# Intelligent Context Override for Titles too
role_lower = cleaned_role.lower()
for key, ctx in ROLE_CONTEXT_MAP.items():
if key in role_lower:
cleaned_role = ctx['synonym'] # Use cleaner role name
break
salary_fmt = ""
if salary_details and salary_details.get("conversions"):
# Prioritize Hourly for Title
if "HOUR" in salary_details["conversions"]:
salary_fmt = salary_details["conversions"]["HOUR"].split('/')[0] + "/hr"
else:
# Fallback to primary
salary_fmt = salary_details.get("primary_display", "").split('/')[0]
parts = {
"role": title_case(cleaned_role),
"company": title_case(company_name),
"city": title_case(city),
"state": state.upper(),
"job_type": emp_types_info.get('title_display', "Part-Time"),
"hiring": random.choice(HIRING_SYNS),
"urgency_tag": job_urgency.get('title_tag', 'Now Hiring'),
"experience_tag": exp_level_info.get('title_tag', ''),
"salary_compact": salary_fmt if "Negotiable" not in salary_fmt else "Competitive Pay",
"skill1": title_case(primary_skill),
"local_area": f"{city} Area"
}
emp_type_key = emp_types_info.get('template_key', 'PART_TIME')
template_list = GEO_FOCUSED_TITLE_TEMPLATES.get(emp_type_key, GEO_FOCUSED_TITLE_TEMPLATES['PART_TIME'])
attempts = 0
tmpl = ""
while attempts < 15:
candidate = random.choice(template_list)
needed = re.findall(r'\{([a-zA-Z_]+)\}', candidate)
# Allow salary_compact to be missing if it's "Competitive Pay" and template strictly wants a number
if "salary_compact" in needed and parts["salary_compact"] == "Competitive Pay":
pass # Some templates might look weird with text, but we'll allow it for variety
missing = [n for n in needed if not parts.get(n)]
if not missing:
tmpl = candidate
break
attempts += 1
if not tmpl: tmpl = "{city} Part-Time: {role}"
title = tmpl.format(**parts)
title = re.sub(r'\s{2,}', ' ', title).strip()
title = re.sub(r'\s*([-|(),:•—])\s*', r'\1', title)
title = title.replace('()', '').replace('[]', '').strip(" -|,: •—")
return enforce_length(title, dynamic_max_len)
def generate_slight_title_improvement(original_title: str, city: str, state: str, urgency_tag: str, max_len: int) -> str:
title = original_title.strip().strip(" -|,: •—")
title_lower = title.lower()
has_city = city and city.lower() in title_lower
has_urgency = urgency_tag and urgency_tag.lower() in title_lower
# High-Visibility "Slight" Templates
template_buckets = {
'urgent_geo': [
"[{urgency_tag}] {title} - {city}",
"{title} ({city}) - {urgency_tag}",
"({urgency_tag}) {title} in {city}"
],
'geo': [
"{title} - {city}",
"{title} [{city}]",
"({city}) {title}",
"{city} Opening: {title}"
],
'urgent': [
"[{urgency_tag}] {title}",
"{urgency_tag}: {title}",
"{title} (Start ASAP)"
],
'bracket_pop': [
"[Part-Time] {title}",
"{title} (PT)",
"{title} [Flexible]"
]
}
parts = {
"title": title,
"city": title_case(city),
"state": state.upper(),
"urgency_tag": urgency_tag,
}
priority_order = []
if urgency_tag and city and not has_city: priority_order.append('urgent_geo')
if city and not has_city: priority_order.append('geo')
if urgency_tag and not has_urgency: priority_order.append('urgent')
priority_order.append('bracket_pop') # Always try to add visual pop if nothing else fits
for bucket_name in priority_order:
templates = template_buckets[bucket_name]
random.shuffle(templates)
for tmpl in templates:
placeholders = re.findall(r'\{([^{}]+)\}', tmpl)
if not all(parts.get(p) for p in placeholders): continue
new_title = tmpl.format(**parts)
if len(new_title) <= max_len:
return enforce_length(new_title, max_len)
return enforce_length(title, max_len)
# ==========================
# Core pipeline
# ==========================
ARGS = None
def rewrite_geotargeted_job_records(in_path: str, out_path: str, seed: int=None, logo_cdn: str = DESIRED_DEFAULT_FALLBACK_LOGO_URL,
default_currency_arg: str = DEFAULT_CURRENCY, enable_salary_adj_arg: bool = False,
full_time_hours_arg: int = DEFAULT_FULL_TIME_HOURS_PER_WEEK,
part_time_hours_arg: int = DEFAULT_PART_TIME_HOURS_PER_WEEK,
no_change_in_title_arg: bool = False,
slightly_improve_title_arg: bool = False):
logging.info(f"Optimizing Geo-Targeted Jobs (Dominator Mode): {in_path} -> {out_path}")
stats = {"total": 0, "title_len_sum": 0}
try:
with open(in_path, 'r', encoding='utf-8') as fin_check:
num_lines = sum(1 for line in fin_check if line.strip())
with open(in_path, 'r', encoding='utf-8') as fin, open(out_path, 'w', encoding='utf-8') as fout:
for line in tqdm(fin, total=num_lines, desc="Dominating Part-Time SEO"):
if not line.strip(): continue
try: rec = json.loads(line)
except json.JSONDecodeError: continue
if not isinstance(rec, dict): continue
jid = rec.get('@id') or rec.get('url') or hashlib.sha256(line.encode()).hexdigest()
random.seed(hash(str(jid) + str(seed)))
# --- BRANCH 1: Slight Improvement Mode ---
if slightly_improve_title_arg:
original_title = rec.get('title', rec.get('name', ''))
if not original_title:
fout.write(line)
continue
city, state, _ = get_location_details(rec)
job_urgency = get_job_urgency_tags(rec.get('datePosted'), rec.get('validThrough'), jid)
dynamic_max_len = MAX_TITLE_LEN_BASE
if stats["total"] >= max(10, num_lines // 2):
avg_title_len = (stats["title_len_sum"] / max(stats["total"], 1))
if avg_title_len < 52: dynamic_max_len = min(84, MAX_TITLE_LEN_BASE + 10)
elif avg_title_len > 72: dynamic_max_len = max(60, MAX_TITLE_LEN_BASE - 5)
new_title = generate_slight_title_improvement(
original_title, city, state,
job_urgency.get('title_tag', ''),
dynamic_max_len
)
rec['title'] = new_title
stats["total"] += 1
fout.write(json.dumps(rec, ensure_ascii=False, sort_keys=True) + "\n")
continue
# --- END BRANCH 1 ---
# --- BRANCH 2: Full Optimization Mode ---
primary_skill = get_primary_skill(rec.get('skills',''))
emp_types_info = get_employment_types_info(rec) # Now uses Aggressive classification
industries_info = get_industries_info(rec)
exp_level_info = get_experience_level_info(rec)
if rec.get('datePosted'):
rec['datePosted'] = to_dhaka_offset(rec['datePosted'])
if rec.get('validThrough'):
rec['validThrough'] = to_midnight(rec['validThrough'])
job_urgency = get_job_urgency_tags(rec.get('datePosted'), rec.get('validThrough'), jid)
curr_code = rec.get('baseSalary', {}).get('currency', default_currency_arg) or default_currency_arg
curr_symbol = get_currency_symbol(curr_code)
salary_details = format_salary_details(
rec, curr_symbol, enable_salary_adj_arg,
emp_types_info['chosen_for_description'],
full_time_hours_arg, part_time_hours_arg
)
rec['description'] = create_geo_targeted_summary(
rec, primary_skill, salary_details, job_urgency,
exp_level_info, industries_info, emp_types_info
) + "
" + (rec.get('description','') or "")
dynamic_max_len = MAX_TITLE_LEN_BASE
if stats["total"] >= max(10, num_lines // 2):
avg_title_len = (stats["title_len_sum"] / max(stats["total"], 1))
if avg_title_len < 52: dynamic_max_len = min(84, MAX_TITLE_LEN_BASE + 10)
elif avg_title_len > 72: dynamic_max_len = max(60, MAX_TITLE_LEN_BASE - 5)
if no_change_in_title_arg:
pass
else:
rec['title'] = generate_location_focused_title(
rec, primary_skill, salary_details, job_urgency,
exp_level_info, emp_types_info, industries_info,
dynamic_max_len
)
# Schema Fixes
rec['employmentType'] = emp_types_info['schema_list'] or None
rec['industry'] = industries_info['schema_list'] or None
if rec.get('url'):
rec['url'] = normalize_url(rec['url'])
else:
slug = re.sub(r'[^a-z0-9]+', '-', rec.get('title', '').lower()).strip('-')
rec['url'] = normalize_url(f"https://walmart.uscareers.co.com/jobs/{slug}/")
ho = rec.get('hiringOrganization')
if isinstance(ho, dict):
rec['hiringOrganization'] = clean_role_and_company(rec.get('title',''), ho.get('name'))[1]
if not ho.get('name'): ho['name'] = "Local Employer"
if not ho.get('logo'): ho['logo'] = logo_cdn
rec['hiringOrganization'] = ho
else:
_, cname = clean_role_and_company(rec.get('title',''), None)
rec['hiringOrganization'] = {"@type":"Organization", "name":cname, "logo": logo_cdn}
bs = rec.get('baseSalary', {})
if isinstance(bs, dict):
bs.setdefault('@type','MonetaryAmount')
bs['currency'] = curr_code
v = bs.get('value', {})
if not isinstance(v, dict): v = {}
v.setdefault('@type','QuantitativeValue')
if salary_details.get("is_negotiable"):
v.update({'description': "Negotiable", 'minValue': None, 'maxValue': None, 'unitText': None})
elif salary_details.get("primary_raw_min") is not None:
v.update({
'minValue': str(salary_details.get("primary_raw_min")),
'maxValue': str(salary_details.get("primary_raw_max")),
'unitText': salary_details.get("primary_unit_normalized", "PROJECT").upper(),
})
bs['value'] = v
rec['baseSalary'] = bs
if rec.get('baseSalary', {}).get('value', {}).get('description') is None:
try: del rec['baseSalary']['value']['description']
except KeyError: pass
rec.update({'@context':'http://schema.org', '@type':'JobPosting'})
for k in list(rec.keys()):
if rec[k] is None: del rec[k]
stats["total"] += 1
stats["title_len_sum"] += len(rec.get('title', ''))
fout.write(json.dumps(rec, ensure_ascii=False, sort_keys=True) + "\n")
if stats["total"]:
avg_t = stats["title_len_sum"] / stats["total"]
logging.info(f"Summary: avg title len={avg_t:.1f}, n={stats['total']}")
except FileNotFoundError:
logging.error(f"Input file '{in_path}' not found.")
except Exception as e:
logging.error(f"Error: {e}", exc_info=True)
logging.info(f"Done. Output: {out_path}")
def main():
global ARGS
ARGS = parse_args()
log_level = logging.DEBUG if ARGS.verbose else logging.INFO
logging.basicConfig(level=log_level, format="%(asctime)s - %(message)s", datefmt="%H:%M:%S")
rewrite_geotargeted_job_records(
ARGS.input, ARGS.output, ARGS.seed, ARGS.logo_cdn, ARGS.currency,
ARGS.enable_salary_adjustment, ARGS.full_time_hours, ARGS.part_time_hours,
ARGS.no_change_in_title, ARGS.slightly_improve_title
)
if __name__ == "__main__":
main()