Azure PurviewでPDFファイルの中身を見て関連する語彙を適用する

# この記事でやること
この記事では、PDFファイルを読んで中身のテキスト情報からAzure Purviewに登録されている語彙にマッチするものを適用していきます。
PDFファイルはADLS Gen2上にあります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/89512/816ef2e6-b00f-c8f0-3b0c-6ead07d003b5.png)


# 準備
PDFファイルは事前にスキャンしておきます。
またPurviewには予め適用する語彙を登録しておきます。
今回は、**PDF分類用Root**の下に登録しています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/89512/3c380b1c-8265-f330-c977-18001e0639c1.png)

# 実装
## PyApacheAtlasの利用
Purviewへのアクセスに[PyApacheAtlas](https://github.com/wjohnson/pyapacheatlas)を使います。
APIの利用開始は[この辺](https://docs.microsoft.com/ja-jp/azure/purview/tutorial-using-rest-apis#create-a-service-principal-application)を参考に。
```python
from pyapacheatlas.core import PurviewClient
from pyapacheatlas.auth import ServicePrincipalAuthentication

auth = ServicePrincipalAuthentication(
    tenant_id = "テナントID", 
    client_id = "クライアントID", 
    client_secret = "シークレット"
)

client = PurviewClient(
    account_name = "Purviewアカウント名",
    authentication = auth
)
```

## 語彙の取得
語彙のdisplayTextとtermGuidを取得します。
今回利用する語彙のルートとして**PDF分類用Root**を登録してあるので配下の語彙だけを取得します。
```python
terms = []
glossary = client.get_glossary()
for term in glossary["terms"]:
    if(term["displayText"].startswith("PDF分類用Root_")):
        terms.append([term["displayText"], term["termGuid"]])
```

## ADLS Gen2上のPDFへのアクセス
ADLS上のファイルへのアクセスは[ここ](https://docs.microsoft.com/ja-jp/azure/storage/blobs/data-lake-storage-directory-file-acl-python)を参照。
PDFからのテキスト抽出は[pdfminer.six](https://github.com/pdfminer/pdfminer.six)を利用。（日本語に対応ということで。）
```python
import json
from azure.storage.filedatalake import DataLakeServiceClient
from pdfminer.high_level import extract_text
from io import BytesIO

storage_account_name = "ストレージ アカウント名"
storage_account_key = "ストレージ アカウント キー"

service_client = DataLakeServiceClient(account_url="{}://{}.dfs.core.windows.net".format(
    "https", storage_account_name), credential=storage_account_key)

file_system="ADLSコンテナ名"
directory="ディレクトリパス"

file_system_client = service_client.get_file_system_client(file_system)
directory_client = file_system_client.get_directory_client(directory)
paths = file_system_client.get_paths(directory)
```
## エンティティと語彙の紐付け
エンティティに語彙を紐付けるには、AtlasGlossarySemanticAssignmentタイプのリレーションを作成します。
```python
relationships = []
for path in paths:
    qualifiedName=f"https://{storage_account_name}.dfs.core.windows.net/{file_system}/{path.name}"
    file_client = directory_client.get_file_client(path)
    download = file_client.download_file()
    downloaded_bytes = download.readall()
    pdf_file = BytesIO(downloaded_bytes)
    text = extract_text(pdf_file, maxpages=1) #サンプルなので1ページだけ

    for term in terms:
        if(term[0].rsplit("_")[1] in text):
            relationship = {
                "typeName": "AtlasGlossarySemanticAssignment",
                "attributes": {},
                "guid": -100,
                "end1": {
                    "guid": term[1]
                },
                "end2": {
                        "typeName": "azure_datalake_gen2_path",
                        "uniqueAttributes": {
                            "qualifiedName": qualifiedName
                        }
                },
            }
            relationships.append(relationship)
```
## 登録
リレーションシップを登録します。
```python
for relationship in relationships:
    entity = relationship["end2"]["uniqueAttributes"]["qualifiedName"]
    print(f"Working on {entity}")

    try:
        results = client.upload_relationship(relationship)
        print("\tSuccess")
    except Exception as e:
        print(f"Exception for {entity} and was not uploaded: {e}")
```

# 応用
今回はテキストに一致する語彙を登録しましたが、機械学習などで分類した結果で関連付けなども可能かと思います。