python-twitter-v2 로 트윗 가져오기

트위터의 내용을 가져오고 싶었습니다. 그래서, 트위터 API를 파 보았습니다.

0. 준비

라이브러리는 python-twitter-v2로 정했습니다.(https://developer.twitter.com/en/portal/dashboard)
예제가 잘 되어 있었고, 쉽고 빠르게 코딩하기 편한 파이썬 라이브러리 였기 때문입니다. 찾으면서 알게 된 것인데 라이브러리가 REST API 를 래핑한 것이어서 없으면 그냥 만들어도 되었 ...

트위터 개발자 대시보드에 들어가서 프로젝트를 생성합니다.

생성할 때만 볼 수 있고 그 후에는 폐지,재생성만 가능하므로 생성할 때 안전한 곳에 잘 기록해두시기 바랍니다.

1. 공개된 트윗을 가져오기

https://pypi.org/project/python-twitter-v2/ 에 인증을 받는 예제를 참고하시면 됩니다.

예제로서 https://twitter.com/currency_bot_kr 에 10분에 한 번씩 올라오는 환율 정보를 가져오려고 하는데, 공개되어 있는 public 트윗이므로 bearer token을 사용하면 됩니다.

인증 & 초기화

>>>> from pytwitter import Api
>>>> api = Api(bearer_token=BEARER_TOKEN)

그리고, get_user 함수를 사용하면 해당 트위터의 User ID를 알아낼 수 있습니다.

>>>> api.get_user(username='currency_bot_kr')
Response(data=User(id='628175003', name='환율봇', username='currency_bot_kr'))

https://tools.codeofaninja.com/find-twitter-id 와 같은 사이트를 이용할 수도 있습니다. 그렇게 해서 해당 계정의 User ID 가 628175003 라는 것을 알게 되었습니다.

>>> tweets = api.get_timelines(user_id="628175003")
>>> tweets
Response(data=[Tweet(id=1537691523378483200, text=2022-06-17 15:58:27 USD: 1293.00 ▼ 0.50...), Tweet(id=1537683970015764480, text=2022-06-17 15:28:53 USD: 1287.60 ▼ 5.90...), Tweet(id=1537676420566364160, text=2022-06-17 14:55:52 USD: 1287.60 ▼ 5.90...), Tweet(id=1537668874082332672, text=2022-06-17 14:29:37 USD: 1286.70 ▼ 6.80...), Tweet(id=1537661320593838080, text=2022-06-17 13:58:39 USD: 1283.50 ▼ 10.00...), Tweet(id=1537653772138475522, text=2022-06-17 13:27:23 USD: 1285.20 ▼ 8.30...), Tweet(id=1537646222827061248, text=2022-06-17 12:55:43 USD: 1288.80 ▼ 4.70...), Tweet(id=1537638673558425600, text=2022-06-17 12:28:48 USD: 1287.20 ▼ 6.30...), Tweet(id=1537631124482306048, text=2022-06-17 11:59:29 USD: 1288.90 ▼ 4.60...), Tweet(id=1537623574760292354, text=2022-06-17 11:29:53 USD: 1289.30 ▼ 4.20...)])
>>> len(tweets.data)
10
>>> tweets.data[0]
Tweet(id=1537691523378483200, text=2022-06-17 15:58:27 USD: 1293.00 ▼ 0.50...)

트위터가 쓰여진 순서대로 가져오므로, 가장 최신의 데이터를 얻고 싶으면 0번째 데이터를 사용하면 됩니다. 얻어오는 트윗의 개수를 max_results 인자로 5에서 100사이 값을 지정할 수 있습니다.

>>> tweets = api.get_timelines(user_id="628175003",max_results=5)
>>> tweets
Response(data=[Tweet(id=1537691523378483200, text=2022-06-17 15:58:27 USD: 1293.00 ▼ 0.50...), Tweet(id=1537683970015764480, text=2022-06-17 15:28:53 USD: 1287.60 ▼ 5.90...), Tweet(id=1537676420566364160, text=2022-06-17 14:55:52 USD: 1287.60 ▼ 5.90...), Tweet(id=1537668874082332672, text=2022-06-17 14:29:37 USD: 1286.70 ▼ 6.80...), Tweet(id=1537661320593838080, text=2022-06-17 13:58:39 USD: 1283.50 ▼ 10.00...)])
>>> len(tweets.data)
5

하루에 가져올 수 있는 트윗의 개수가 정해져 있으므로 max_results 값을 적절하게 쓰는 것도 좋을 것입니다.

>>> tweets.meta
Meta(result_count=5)
>>> tweets.meta.next_token
'7140dibdnow9c7btw421tf8ra9mpcyf73gwp9f66mlnhs'

한 번에 100개 까지 가져올 수 있다고 했습니다. 이 경우에는 필요가 없지만, 만일 100개 이상의 트윗을 가져오고 싶을 때는 next_token을 사용해서 다음 것을 이어서 가져올 수 있습니다. 트위터 개발자 페이지의 설명에 따르면, 이 토큰의 유효기간은 없다고 합니다.

2. 중복 처리 및 데이터 저장을 위해 sqlite 사용하기

최근 트윗을 최소 5개 가져와야 하므로 중복이 생길 수 밖에 없습니다. 따라서, 중복되는 것의 처리와 데이터 저장을 위해 sqlite3를 사용하겠습니다.

먼저 테이블이 없을 때 생성하는 함수, 그리고 트윗을 저장하는 두 개의 함수를 만듭니다.

import sqlite3

def createTableIfNotExist(con):
    try:
        sqlcmd = "CREATE TABLE IF NOT EXISTS data (id VARCHAR(20) unique, text TEXT)"
        con.execute(sqlcmd)
        con.commit()
    except Exception as e:
        print(e)
        return False
    return True

def insertTweetToTable(con, id, text):
    try:
        con.execute("INSERT INTO data VALUES(?,?)", (id, text))
    except Exception as e:
        print(e)
        return False

    return True

트윗을 얻어온 후 db 파일에 연결해서 위의 함수를 이용해서 테이블에 기록하면 됩니다.

BEARER_TOKEN = "<Your bearer token>"
TWITTER_ACCOUNT ='currency_bot_kr'
MAX_RESULT = 5 # 최대값은 100 입니다.

api = Api(bearer_token=BEARER_TOKEN)
userinfo = api.get_user(username=TWITTER_ACCOUNT)
userid = userinfo.data.id
tweets = api.get_timelines(user_id=userid,max_results=5)

print(tweets)

dbpath = 'tweet.db'
try:
    con = sqlite3.connect(dbpath)
    if createTableIfNotExist(con) == False:
        con.close()
        con = None

    for data in tweets.data:
        insertTweetToTable(con, data.id, data.text)
        print(data.id, data.text)

    con.commit()

except Exception as e:
    print(e)

con.close()

만일 같은 트윗이 DB에 기록이 된다면 'UNIQUE constraint failed:' 예외가 발생되고 중복 기록을 방지하며, 트윗 ID는 시간 순서에 따르므로 가장 오래된 것과 최신의 것을 쉽게 알 수 있을 것입니다.

이제 DB에 있는 내용을 잘 요리하시면 됩니다.

3. sqlite3 에 대해 알게된 의외의 사실

하나의 데이터베이스를 여러 프로그램(또는 프로세서)이 동시에 읽고 쓰는 경우 - 예를 들어 DB에 가져온 것을 파이썬 웹 서버를 이용해서 웹에 보여주고 싶다거나, 여러 트윗을 하나의 DB 에 같이 저장하고 싶을 때는 sqlite3가 아닌 MySQL, MariaDB 같은 것을 써야 한다고 생각했었습니다.

하지만, sqlite3 에서 가능한 것으로 보입니다. (MySQL 같은 DB엔진보다 성능은 떨어지겠지만)

sqlite3 의 FAQ(https://sqlite.org/faq.html#q5) 에도 설명이 되어 있는데 "여러 프로세서가 동시에 동일한 데이터 베이스를 열 수 있고, 동시에 SELECT를 수행할 수 있지만, 어떤 순간에는 단 하나의 프로세서 만이 데이터 베이스를 변경할 수 있습니다" 라는 내용이 있습니다.

그리고, 관련해서 누군가 테스트한 파이썬 코드가 올려진 깃헙도 찾았습니다. (https://github.com/joedougherty/sqlite3_concurrent_writes_test_suite)

sqlite 를 복제해주는 Litestream 과 같은 프로그램도 있습니다.