汎用的な奴がほしかったので。
文字コードを判定するようにしています。
内向きWebhookを使う版
import csv
import requests
import chardet
def read_csv(file_path):
with open(file_path, "rb") as f:
result = chardet.detect(f.read())
data = []
with open(file_path, newline='', encoding=result["encoding"]) as csvfile:
csvreader = csv.reader(csvfile)
for row in csvreader:
data.append(row)
return data
def convert_to_markdown_table(data):
headers = data[0]
table = ["| " + " | ".join(headers) + " |"]
table.append("| " + " | ".join(["---"] * len(headers)) + " |")
for row in data[1:]:
table.append("| " + " | ".join(row) + " |")
return "\n".join(table)
def post_to_mattermost(table, config):
webhook_url = config["webhook_url"]
payload = {
"text": table
}
response = requests.post(webhook_url, json=payload)
if response.status_code != 200:
raise Exception("Failed to post message")
if __name__ == "__main__":
input_csv = "input.csv" # CSVファイルのパス
mattermost_config = {
"webhook_url": "your-incoming-webhook-url", # 内向きWebhookのURL
}
csv_data = read_csv(input_csv)
markdown_table = convert_to_markdown_table(csv_data)
post_to_mattermost(markdown_table, mattermost_config)
APIキーを使う版
import csv
import requests
import chardet
def read_csv(file_path):
with open(file_path, "rb") as f:
result = chardet.detect(f.read())
data = []
with open(file_path, newline='', encoding=result["encoding"]) as csvfile:
csvreader = csv.reader(csvfile)
for row in csvreader:
data.append(row)
return data
def convert_to_markdown_table(data):
headers = data[0]
table = ["| " + " | ".join(headers) + " |"]
table.append("| " + " | ".join(["---"] * len(headers)) + " |")
for row in data[1:]:
table.append("| " + " | ".join(row) + " |")
return "\n".join(table)
def post_to_mattermost(table, config):
base_url = config["url"]
api_key = config["api_key"]
team_name = config["team_name"]
channel_name = config["channel_name"]
headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}
# Get team and channel IDs
team_url = f"{base_url}/api/v4/teams/name/{team_name}"
channel_url = f"{base_url}/api/v4/teams/name/{team_name}/channels/name/{channel_name}"
response = requests.get(team_url, headers=headers)
team_id = response.json()["id"]
response = requests.get(channel_url, headers=headers)
channel_id = response.json()["id"]
# Create post
post_url = f"{base_url}/api/v4/posts"
post_data = {
"channel_id": channel_id,
"message": table
}
response = requests.post(post_url, headers=headers, json=post_data)
if response.status_code != 201:
raise Exception("Failed to post message")
if __name__ == "__main__":
input_csv = "input.csv" # CSVファイルのパス
mattermost_config = {
"url": "your-mattermost-url", # MattermostサーバーのURL
"api_key": "your-api-key", # あなたのAPIキー
"team_name": "your-team-name", # チーム名
"channel_name": "your-channel-name", # チャンネル名
}
csv_data = read_csv(input_csv)
markdown_table = convert_to_markdown_table(csv_data)
post_to_mattermost(markdown_table, mattermost_config)