XML parser + bs4 combo

Question

I just wrote a short function to parse SO jobs XML feed and return dicts containing information about each job entry.
Now, each job entry page is visited to grab info not present in the xml feed (company logo URL, salary info, etc.) through bs4 with couple functions in the get_so_extras module.

parse_feed.py:

from datetime import datetime as dt
from xml.etree import ElementTree as etree
from get_so_extras import get_company_logo, get_so_salary
import urllib3
import json
import sys


def get_so_listings(url):
    """
    Get remote job listings from stackoverflow.com xml feed
    """

    http = urllib3.PoolManager()

    try:
        so_file = http.request("GET", url).data
    except Exception as e:
        print(e.args)
        sys.exit()

    so_root = etree.fromstring(so_file)
    items = so_root.findall("channel/item")

    # Parse SO xml feed in dict
    so_listings = []
    for e in items:
        job_dict = {
            "so_id": e.findtext("guid"),
            "title": e.findtext("title"),
            "so_url": e.findtext("link"),
            "author": e.findtext(
                """{http://www.w3.org/2005/Atom}author/{http://www.w3.org/2005/Atom}name"""
            ),
            "description": e.findtext("description"),
            "published_at": str(
                dt.strptime(
                    e.findtext("pubDate").strip(), "%a, %d %b %Y %H:%M:%S Z"
                    )
            ),
            "updated_at": str(
                dt.strptime(
                    e.findtext("{http://www.w3.org/2005/Atom}updated"),
                    "%Y-%m-%dT%H:%M:%SZ",
                    )
            ),
            "company_logo_url": get_company_logo(e.findtext("link")),
            "categories": [c.text for c in e.findall("category")],
        }

        # @TODO: get_company_logo(job_dict["so_url"])
        so_listings.append(job_dict)

    print("Jobs imported: {}".format(len(so_listings)))

    # Helper - export to json, ideally pushed to db
    with open('output1.json', 'w', encoding='utf-8') as f:
        json.dump(so_listings, f, indent=4)

    return True

And here the helper function used to grab company logo URLs from get_so_extras module.

get_so_extras.py:

from bs4 import BeautifulSoup
import requests
import time


def get_company_logo(job_url):
    """
    Get company logo from stackoverflow.com job listing page
    """

    try:
        page = requests.get(job_url)
        soup = BeautifulSoup(page.text, 'html.parser')
        logo = soup.find("div", attrs={'class': 's-avatar s-avatar__lg mr8 bg-white fl-shrink0'}).\
            img["src"]
    except Exception as e:
        print(e)
        pass

    time.sleep(5) # Be kind

    return logo

It'd be great to understand whether I made any obvious mistake or bad choice along the way. Obviously calling get_company_logo within the XML parsing loop slows things down, what would be the best approach to get those fields?
Probably checking if we're looking at a new company/listing from db before visiting the page to scrape the logo would be more efficient.

edit: also, given job listings have pretty much similar attributes, could it make sense to create a class listing with attribute and a method to push it to db (with de-duplication logic perhaps)?

Be ruthless, thanks in advance.

Answer 1

I've been experimenting with the best data structure for this. Here is what I have for now. It is definitely a work in progress, I hope to keep updating it constantly.

The CSV output was unexpected, ultimately a result of my decision to use a namedtuple. I chose namedtuples because, like you, I thought the data fit that tabular style quite well. You wrote given job listings have pretty much similar attributes, could it make sense to create a class listing with attribute and a method to push it to db, and I mostly agree, I just didn't feel that the data itself warranted a whole class.

import collections as colls
import csv
import datetime as dt

import requests
from lxml import etree

Job = colls.namedtuple('Job', ['guid', 'title', 'url', 'author', 'logo', 'description', 'pub_date', 'update_date',
                               'categories'])

html_parser = etree.HTMLParser()

req_session = requests.Session()
req = req_session.get(url_1)

root = etree.fromstring(req.content)
ns_map = root.nsmap

job_elems = root.xpath('/rss/channel/item')[:10]


def parse_job_item(job_item):
    guid = job_item.findtext('guid')
    title = job_item.findtext('title')
    link = job_item.findtext('link')
    author = job_item.findtext('a10:author/a10:name', namespaces=ns_map)
    description = job_item.findtext('description')
    pub_date = job_item.findtext('pubDate')
    if pub_date:
        pub_date = dt.datetime.strptime(pub_date, "%a, %d %b %Y %H:%M:%S Z")
    update_date = job_item.findtext('a10:updated', namespaces=ns_map)
    if update_date:
        update_date = dt.datetime.strptime(update_date, "%Y-%m-%dT%H:%M:%SZ")
    categories = [elem.text for elem in job_item.findall('category')]

    job_page_elem = etree.fromstring(req_session.get(link).content, parser=html_parser)
    # "//body/div[@class='container']/div[@id='content']/header/div/a/img/@src"
    # contains(concat(' ', normalize-space(@class), ' '), ' s-avatar ')
    company_logo = job_page_elem.xpath(
        "/html/body/div[@class='container']/div[@id='content']/header/div[contains(concat(' ', normalize-space(@class), ' '), ' s-avatar ')]//img/@src")
    if company_logo:
        company_logo = company_logo[0]
    res_item = Job(guid, title, link, author, company_logo, description, pub_date, update_date, categories)
    return res_item


job_data_list = (parse_job_item(curr_elem) for curr_elem in job_elems)

with open('../out/jobs_out.csv', 'w', newline='') as out_file:
    writer = csv.writer(out_file)
    writer.writerow(Job._fields)
    writer.writerows(job_data_list)