Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_bad_json(self):
with self.assertRaises(Exception) as context:
markovify.Chain.from_json(1)
def test_bad_corpus(self):
with self.assertRaises(Exception) as context:
markovify.Chain(corpus="testing, testing", state_size=2)
def from_cache():
#log("%s: Reading cache for %s" % (id, user))
f_txt = open(txt_fname, 'r')
f_json = open(json_fname, 'r')
text = ''.join(f_txt.readlines()).split(HIST_DEL)
json = f_json.readlines()[0]
if text == [] or json == []:
return from_scratch()
f_txt.close()
f_json.close()
return QText(text, state_size=STATE_SIZE, chain=markovify.Chain.from_json(json))
log("%s: Using cache for %s" % (id, source), console_only=True)
f_txt = open(txt_fname, 'r')
f_json = open(json_fname, 'r')
f_info = open(info_fname, 'r')
text = ''.join(f_txt.readlines())
json = f_json.readlines()[0]
try:
sentence_avg = int(f_info.readlines()[0])
except ValueError:
sentence_avg = 1
if text == '' or json == []:
return from_scratch()
f_txt.close()
f_json.close()
f_info.close()
return (PText(text, state_size=STATE_SIZE, chain=markovify.Chain.from_json(json)), sentence_avg)
def __init__(self, do_markovify=False):
self.tagged_sents = list(brown.tagged_sents())
self.model = None
if do_markovify:
self.model = markovify.Chain(self.tagged_sents, 2)
self.words_from_pos = dict()
for word, tag in brown.tagged_words():
self.words_from_pos.setdefault(tag, []).append(word)
with open(database_path('settings.yaml')) as f:
self.startswith = yaml.safe_load(f)['startswith']['brown']
def __init__(self, input_text, state_size=2, chain=None):
"""
input_text: A list of strings representing individual comments.
state_size: An integer indicating the number of words in the model's state.
chain: A trained markovify.Chain instance for this text, if pre-processed.
"""
if chain == None:
runs = self.generate_corpus(input_text)
self.input_text = input_text
self.state_size = state_size
self.chain = chain or markovify.Chain(runs, state_size)
raise ValueError('invalid type for chat')
if not message:
raise ValueError('message cannot be empty')
logger.debug(f'updating model for chat-id:{chat.id}')
model = new_model(message)
if not model:
return
else:
model = model.chain
chat_id = str(chat.id)
chat_data = db.find_one(chat_id=chat_id) or {}
text = '\n'.join([chat_data.get('text', ''), message])
if chat_data.get('chain', ''):
if settings.GROW_CHAIN:
chain = json.loads(chat_data.get('chain'))
cur_m = markovify.Chain.from_json(chain)
model = markovify.combine([cur_m, model])
else:
text = '\n'.join(
text.splitlines()[-settings.MESSAGE_LIMIT:])
model = new_model(text).chain
db.upsert({
'chat_id': chat_id,
'text': text,
'chain': json.dumps(model.to_json())
}, ['chat_id'])