Takes POSTed data and determines the format, and returns an Edition record suitable for adding to OL. :param bytes data: Raw data :return: (Edition record, format (rdf|opds|marcxml|json|marc)) or (None, None)
(data: bytes)
| 64 | |
| 65 | |
| 66 | def parse_data(data: bytes) -> tuple[dict | None, str | None]: |
| 67 | """ |
| 68 | Takes POSTed data and determines the format, and returns an Edition record |
| 69 | suitable for adding to OL. |
| 70 | |
| 71 | :param bytes data: Raw data |
| 72 | :return: (Edition record, format (rdf|opds|marcxml|json|marc)) or (None, None) |
| 73 | """ |
| 74 | data = data.strip() |
| 75 | if b"<?xml" in data[:10]: |
| 76 | root = etree.fromstring(data, parser=lxml.etree.XMLParser(resolve_entities=False)) |
| 77 | if root.tag == "{http://www.w3.org/1999/02/22-rdf-syntax-ns#}RDF": |
| 78 | edition_builder = import_rdf.parse(root) |
| 79 | format = "rdf" |
| 80 | elif root.tag == "{http://www.w3.org/2005/Atom}entry": |
| 81 | edition_builder = import_opds.parse(root) |
| 82 | format = "opds" |
| 83 | elif root.tag == "{http://www.loc.gov/MARC21/slim}record": |
| 84 | if root.tag == "{http://www.loc.gov/MARC21/slim}collection": |
| 85 | root = root[0] |
| 86 | rec = MarcXml(root) |
| 87 | edition = read_edition(rec) |
| 88 | edition_builder = import_edition_builder.import_edition_builder(init_dict=edition) |
| 89 | format = "marcxml" |
| 90 | else: |
| 91 | raise DataError("unrecognized-XML-format") |
| 92 | elif data.startswith(b"{") and data.endswith(b"}"): |
| 93 | obj = json.loads(data) |
| 94 | |
| 95 | # Only look to the import_item table if a record is incomplete. |
| 96 | # This is the minimum to achieve a complete record. See: |
| 97 | # https://github.com/internetarchive/openlibrary/issues/9440 |
| 98 | # import_validator().validate() requires more fields. |
| 99 | minimum_complete_fields = ["title", "authors", "publish_date"] |
| 100 | is_complete = all(obj.get(field) for field in minimum_complete_fields) |
| 101 | if not is_complete: |
| 102 | isbn_10 = safeget(lambda: obj.get("isbn_10", [])[0]) |
| 103 | isbn_13 = safeget(lambda: obj.get("isbn_13", [])[0]) |
| 104 | identifier = to_isbn_13(isbn_13 or isbn_10 or "") |
| 105 | |
| 106 | if not identifier: |
| 107 | identifier = get_non_isbn_asin(rec=obj) |
| 108 | |
| 109 | if identifier: |
| 110 | supplement_rec_with_import_item_metadata(rec=obj, identifier=identifier) |
| 111 | |
| 112 | edition_builder = import_edition_builder.import_edition_builder(init_dict=obj) |
| 113 | format = "json" |
| 114 | elif data[:MARC_LENGTH_POS].isdigit(): |
| 115 | # Marc Binary |
| 116 | if len(data) < MARC_LENGTH_POS or len(data) != int(data[:MARC_LENGTH_POS]): |
| 117 | raise DataError("no-marc-record") |
| 118 | record = MarcBinary(data) |
| 119 | edition = read_edition(record) |
| 120 | edition_builder = import_edition_builder.import_edition_builder(init_dict=edition) |
| 121 | format = "marc" |
| 122 | else: |
| 123 | raise DataError("unrecognised-import-format") |
no test coverage detected