Проблема при записи извлеченной информации из PDF-файлов в CSV-файл - PullRequest
Новая
       8

Проблема при записи извлеченной информации из PDF-файлов в CSV-файл

0 голосов
/ 17 июня 2020

Интересно, есть ли у кого-нибудь шанс помочь мне решить следующую проблему. Я написал функцию, пытающуюся извлечь информацию из файлов PDF, а затем я хочу поместить всю полученную информацию в файл csv. Основываясь на том, что я нашел в Google, я также написал другую функцию для хранения результатов и попытался использовать командную строку для простого вызова.

Но когда я попытался запустить следующий код, мой результат csv-файл был всегда пусто без ввода данных. Я как бы запуталась в этом сценарии и задаюсь вопросом, может ли кто-нибудь помочь. 

from collections import OrderedDict
from csv import DictWriter
from glob import glob
import os
from pathlib import Path
import re
import string
import typing

import click
import dateparser
# pylint: disable=no-name-in-module
from lxml.etree import _ElementTree
from pdfminer.pdfparser import PDFSyntaxError
import pdfquery
from pdfquery.cache import FileCache
from pdfquery.pdfquery import PDFQuery
from tqdm import tqdm

slicer = "#" if os.name == "nt" else "-"

def parse_date(date):
    return dateparser.parse(date, languages=["en"]).strftime(f"%{slicer}m/%{slicer}d/%y")

def extract_newsbank(pdf: PDFQuery) -> list:
    # Create an ordered dictionary to store the title
    titles = OrderedDict.fromkeys(title.getparent() if title.text else title for title in pdf.pq("LTTextLineHorizontal[height='16.896']"))
    # Create a new list to store articles
    articles = {}

    for title in titles:
        for parent in title.iterancestors():
            index = parent.get("page_index")
            if index:
                articles[int(index)] = {"Headline": "".join(title.itertext()).strip(), "Author": "", "Section": ""}

    start_pages = list(articles.keys())
    extra = start_pages + [int(pdf.pq("LTPage")[-1].get("pageid"))]

    for i, page in tqdm(list(enumerate(start_pages))):
        article = articles[page]
        pages = pdf.get_tree(*range(page, extra[i + 1]))
        if type(pages) == _ElementTree:
            pages = pages.getroot()

        details_box = pages.cssselect("LTTextBoxHorizontal[index='1']")[0]
        detail_lines = [elem.text for elem in details_box.cssselect("LTTextLineHorizontal")]

        article["Date"] = parse_date(detail_lines[0].split(" |")[0])

        publication_text = ""
        for line in detail_lines:
            if line.find("Author:") != -1 or line.find("Section:") != -1:
                try:
                    split = line.split(" ")
                    article["Words"] = int(split[-3])
                    line = " ".join(split[:-3])
                except ValueError:
                    pass

                last_group = ""

                for group in re.split("(Section|Author): ", line):
                    if last_group in ["Author", "Section"]:
                        article[last_group] = string.capwords(group.strip())
                    last_group = group

                break

            publication_text += line.split(" | ")[-1]

        article["Publication"] = publication_text.strip()

        boxes = pages.cssselect("LTTextBoxHorizontal")[2:]
        all_text = ""

        for i in range(0, len(boxes)):
            if boxes[i].attrib["height"] == "7.68":
                boxes = sorted(boxes[i + 2:], key=lambda b: float(b.attrib["y0"]), reverse=True)
                break

            all_text += boxes[i].text

        article["News"] = all_text.strip()
        article["Citation"] = "".join([box.text for box in boxes]).strip()

        if "Words" not in article:
            article["Words"] = len(article["News"].split(" "))

    return articles.values()

def extract_articles(files, path):
    """Extract all news articles from FILES and write them to a CSV at PATH."""

    os.makedirs("cache", exist_ok=True)

    all_files = []

    for file in files:
        all_files += glob(file)

    with open(path, "w", newline="\n", encoding="utf-8") as articles_file:
        articles_csv = DictWriter(articles_file, fieldnames=["Date", "Publication", "Section", "Author", "Words", "Headline", "News", "Citation"])
        articles_csv.writeheader()

        for file in all_files:
            print("[*] Processing", file)
            try:
                # Create PDF cache and load from it as a workaround for a bug in pdfquery :(
                # https://github.com/jcushman/pdfquery/issues/53
                pdf = pdfquery.PDFQuery(file, parse_tree_cacher=FileCache("cache/"))
                pdf.load()
                pdf = pdfquery.PDFQuery(file, parse_tree_cacher=FileCache("cache/"))
                pdf.load()

                tree = pdf.tree
                if type(tree) == _ElementTree:
                    tree = tree.getroot()

                producer = tree.attrib["Producer"]

                if "Qt" in producer:
                    articles = extract_newsbank(pdf)
            except PDFSyntaxError:
                print("[!]", file, "is not a PDF file!")
                continue

            articles_csv.writerows(articles)

    for cache_file in glob("cache/pdfquery_*.xml.zip"):
        os.remove(cache_file)
    os.rmdir("cache")

    return

@click.command()
@click.argument("files", required=True, nargs=-1)
@click.argument("path", type=click.Path())
def run_extract_articles(files, path):
    extract_articles(files, path)

if __name__ == "__main__":
    # pylint: disable=no-value-for-parameter
    #run_extract_articles()
    extract_articles(["Campaign_Collects_Names_of_Mur.pdf"], "gale.csv")```
...