#!/usr/bin/env python3 """ rowjobs.site rowjobs Dependencies: • jinja2 (install via pip install jinja2) • BeautifulSoup (install via pip install beautifulsoup4) • Python built-ins: re, random, datetime, argparse, os, json """ import os import json import re import random import datetime import argparse from jinja2 import Template from bs4 import BeautifulSoup # Define the main domain for constructing job URLs. MAIN_DOMAIN = "https://rowjobs.site.com" # Global set to track used random suffixes. used_suffixes = set() def generate_random_suffix(): """Generate a unique 6-digit random number as a string.""" global used_suffixes candidate = ''.join(random.choices('0123456789', k=6)) while candidate in used_suffixes: candidate = ''.join(random.choices('0123456789', k=6)) used_suffixes.add(candidate) return candidate def slugify(title): """ Create a URL-friendly slug from the job title. Converts to lower-case, replaces spaces with hyphens, and removes unwanted characters. """ slug = title.lower() slug = re.sub(r'[^a-z0-9\s-]', '', slug) slug = re.sub(r'\s+', '-', slug).strip('-') return slug def clean_visible_body(html): """ Clean the job body HTML by: • Removing all hyperlink tags ( and ) while preserving their inner text. • Unwrapping all tags (i.e. remove the span tags but keep their inner content). • Removing any class attributes from all remaining tags. """ soup = BeautifulSoup(html, "html.parser") # Remove hyperlink tags but preserve their inner text. for a_tag in soup.find_all('a'): a_tag.unwrap() # Unwrap all span tags (do not decompose, so inner text remains) for span in soup.find_all('span'): span.unwrap() # Remove class attributes from all tags. for tag in soup.find_all(True): tag.attrs.pop('class', None) return str(soup) def ensure_html_linebreaks(text): """ If the cleaned text does not contain HTML tags, replace newline characters with
tags and also convert \u0095 to an HTML bullet. """ if "<" not in text: # No HTML tags found, assume plain text. text = text.replace("\u0095", "•") text = text.replace("\n", "
") return text def remove_html_tags(html): """ Remove all HTML tags from a string, but first convert
and

tags to newline characters. """ # Convert
tags to newline characters. html = re.sub(r'(?i)', '\n', html) # Convert closing

tags to newline characters. html = re.sub(r'(?i)

', '\n', html) # Remove all remaining HTML tags. return re.sub(r'<[^>]+>', '', html) def extract_salary(detail): """ Extract a single numeric salary value from a string (e.g., '$25 per hour' -> '25'). """ match = re.search(r'\$([\d,\.]+)', detail) if match: return match.group(1).replace(',', '') return None def extract_salary_range(detail): """ Extract a salary range from a string. Expected formats: "$18-$25 per hour" or "$360K–$920K a year". Returns a tuple: (min_salary, max_salary, unit) or None. """ pattern = r'\$([\d,\.]+)\s*(K?)[-–]\s*\$([\d,\.]+)\s*(K?)\s*(?:per\s+([\w/]+))?' m = re.search(pattern, detail, re.IGNORECASE) if m: lower_str, lower_mult, upper_str, upper_mult, unit = m.groups() def convert(num_str, multiplier): num_str = num_str.replace(',', '') try: value = float(num_str) except ValueError: value = 0 if multiplier.upper() == 'K': value *= 1000 return int(value) lower_val = convert(lower_str, lower_mult) upper_val = convert(upper_str, upper_mult) unit = unit.upper() if unit else "" return (lower_val, upper_val, unit) return None def format_employment_type(emp_type): """ Convert an employment type string like "FULL_TIME, PART_TIME" into a more human-friendly version. """ parts = [p.strip() for p in emp_type.split(",")] return ", ".join(part.replace("_", " ").title() for part in parts) # HTML template. template_html = r""" {{ job_title }}

Posted {{ date_posted }}

{{ location }} {{ employment_type_display }} Immediate Start

Apply Now

Similar Jobs

Recent Jobs

{% for job in recent_jobs %}

{% endfor %}

""" def process_job(job_record, force_remote=False, force_company=None, force_apply_url=None): """ Process a single job record: • Generate an initial job URL from the job title. • Join the body parts preserving the original HTML. • Clean the job body by removing tags (and unwrapping tags and removing class attributes). • Create a plain text version for the schema description (converting
and

tags to newlines). • Extract salary data. • Apply any forced overrides (location, company, apply URL). • Create an OG description limited to the first 120 words. """ job_title = job_record.get("job_title", "No Title") new_job_url = f"{MAIN_DOMAIN}/job/{slugify(job_title)}" body_parts = job_record.get("body_parts", []) raw_body = "".join(body_parts) visible_body = clean_visible_body(raw_body) # Ensure that if the visible body is plain text, we convert newlines to
tags visible_body = ensure_html_linebreaks(visible_body) plain_description = remove_html_tags(visible_body) truncated_og_description = " ".join(plain_description.split()[:120]) company_name = job_record.get("company_name", "Unknown Company") if force_company: company_name = force_company location = job_record.get("location", "Unknown location") if force_remote: location = "Remote, USA" job_details = job_record.get("job_details", []) today = datetime.date.today() date_posted = today.isoformat() valid_through = (today + datetime.timedelta(days=300)).isoformat() full_time_pattern = re.compile(r'full[-_\s]*time', re.IGNORECASE) part_time_pattern = re.compile(r'part[-_\s]*time', re.IGNORECASE) employment_set = set() for detail in job_details: if full_time_pattern.search(detail): employment_set.add("FULL_TIME") if part_time_pattern.search(detail): employment_set.add("PART_TIME") if not employment_set: employment_set.add("FULL_TIME") employment_type = ", ".join(sorted(employment_set)) employment_type_display = format_employment_type(employment_type) job_location_type = "ONSITE" if "remote" in job_record.get("location", "").lower(): job_location_type = "TELECOMMUTE" if force_remote: job_location_type = "TELECOMMUTE" applicant_country = "US" company_website = MAIN_DOMAIN apply_url = force_apply_url if force_apply_url else new_job_url salary_min = None salary_max = None salary_value = None salary_unit = None for detail in job_details: if "$" in detail and ('-' in detail or '–' in detail): salary_range = extract_salary_range(detail) if salary_range: salary_min, salary_max, unit_extracted = salary_range salary_unit = unit_extracted if unit_extracted else ("HOUR" if "hour" in detail.lower() else "YEAR") break if salary_min is None or salary_max is None: for detail in job_details: if "$" in detail: extracted = extract_salary(detail) if extracted: salary_value = extracted if "hour" in detail.lower(): salary_unit = "HOUR" elif "year" in detail.lower(): salary_unit = "YEAR" else: salary_unit = "MONTHLY" break salary_data = {} if salary_min is not None and salary_max is not None: salary_data["salary_min"] = salary_min salary_data["salary_max"] = salary_max salary_data["salary_unit"] = salary_unit elif salary_value is not None: salary_data["salary_value"] = salary_value salary_data["salary_unit"] = salary_unit else: salary_data["salary_value"] = "Not Disclosed by Recruiter" salary_data["salary_unit"] = "MONTHLY" return { "job_title": job_title, "job_url": new_job_url, "apply_url": apply_url, "job_body": visible_body, "job_description": plain_description, "og_description": truncated_og_description, "company_name": company_name, "location": location, "date_posted": date_posted, "valid_through": valid_through, "employment_type": employment_type, "employment_type_display": employment_type_display, "currency": "USD", "applicant_country": applicant_country, "job_location_type": job_location_type, "company_website": company_website, **salary_data } def generate_html(job_data, template_str): """Render the HTML content using the provided template and job data.""" template = Template(template_str) return template.render(**job_data) def main(ndjson_file, output_dir, force_company, force_apply_filepath): force_apply_url = None if force_apply_filepath: with open(force_apply_filepath, "r", encoding="utf-8") as f: force_apply_url = f.read().strip() if not os.path.exists(output_dir): os.makedirs(output_dir) all_job_data = [] with open(ndjson_file, "r", encoding="utf-8") as f: for idx, line in enumerate(f, start=1): line = line.strip() if not line: continue try: job_record = json.loads(line) # Set force_remote to True by default job_data = process_job(job_record, force_remote=True, force_company=force_company, force_apply_url=force_apply_url) all_job_data.append(job_data) except Exception as e: print(f"Error processing line {idx}: {e}") for job in all_job_data: file_identifier = generate_random_suffix() job["file_identifier"] = file_identifier job["job_url"] = f"{MAIN_DOMAIN}/job/{file_identifier}" if not force_apply_url: job["apply_url"] = job["job_url"] for current_job in all_job_data: other_jobs = [job for job in all_job_data if job is not current_job] recent_jobs = random.sample(other_jobs, min(10, len(other_jobs))) if other_jobs else [] remaining_candidates = [job for job in other_jobs if job not in recent_jobs] suggested_jobs = random.sample(remaining_candidates, min(10, len(remaining_candidates))) if remaining_candidates else [] current_job["recent_jobs"] = recent_jobs current_job["suggested_jobs"] = suggested_jobs file_name = f"{current_job['file_identifier']}.html" output_path = os.path.join(output_dir, file_name) html_content = generate_html(current_job, template_html) with open(output_path, "w", encoding="utf-8") as out_f: out_f.write(html_content) print(f"Generated: {output_path}") total_jobs = len(all_job_data) full_time_only = sum(1 for job in all_job_data if job["employment_type"] == "FULL_TIME") part_time_only = sum(1 for job in all_job_data if job["employment_type"] == "PART_TIME") both = sum(1 for job in all_job_data if job["employment_type"] == "FULL_TIME, PART_TIME") stats_output_path = os.path.join(output_dir, "job_statistics.txt") with open(stats_output_path, "w", encoding="utf-8") as stat_file: stat_file.write(f"Total jobs processed: {total_jobs}\n") stat_file.write(f"Full Time only: {full_time_only}\n") stat_file.write(f"Part Time only: {part_time_only}\n") stat_file.write(f"Both Full Time and Part Time: {both}\n") print(f"Statistics file generated: {stats_output_path}") if __name__ == "__main__": parser = argparse.ArgumentParser( description="Generate HTML files with job posting schema from NDJSON data using a refined template. All jobs are marked as TELECOMMUTE by default. Use --force-company to override the company name, and --force-apply to use a forced apply URL from a file." ) parser.add_argument("ndjson_file", help="Path to the NDJSON file containing job records") parser.add_argument("output_dir", help="Directory for the generated HTML files") parser.add_argument("--force-company", type=str, default=None, help="Force the company name in the schema to this value") parser.add_argument("--force-apply", type=str, default=None, help="Path to a text file containing the forced apply URL") args = parser.parse_args() main(args.ndjson_file, args.output_dir, args.force_company, args.force_apply)

{{ job_title }}

Similar Jobs

Recent Jobs

You May Also Like