<commit_before>package nxcore

import (
	"encoding/json"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/jaracil/ei"
	"github.com/jaracil/smartio"
	"golang.org/x/net/context"
)

const (
	ErrParse            = -32700
	ErrInvalidRequest   = -32600
	ErrMethodNotFound   = -32601
	ErrInvalidParams    = -32602
	ErrInternal         = -32603
	ErrTimeout          = -32000
	ErrCancel           = -32001
	ErrInvalidTask      = -32002
	ErrInvalidPipe      = -32003
	ErrInvalidUser      = -32004
	ErrUserExists       = -32005
	ErrPermissionDenied = -32010
	ErrTtlExpired       = -32011
)

var ErrStr = map[int]string{
	ErrParse:            "Parse error",
	ErrInvalidRequest:   "Invalid request",
	ErrMethodNotFound:   "Method not found",
	ErrInvalidParams:    "Invalid params",
	ErrInternal:         "Internal error",
	ErrTimeout:          "Timeout",
	ErrCancel:           "Cancel",
	ErrInvalidTask:      "Invalid task",
	ErrInvalidPipe:      "Invalid pipe",
	ErrInvalidUser:      "Invalid user",
	ErrUserExists:       "User already exists",
	ErrPermissionDenied: "Permission denied",
	ErrTtlExpired:       "TTL expired",
}

type JsonRpcErr struct {
	Cod  int         `json:"code"`
	Mess string      `json:"message"`
	Dat  interface{} `json:"data,omitempty"`
}

func (e *JsonRpcErr) Error() string {
	return fmt.Sprintf("[%d] %s", e.Cod, e.Mess)
}

func (e *JsonRpcErr) Code() int {
	return e.Cod
}

func (e *JsonRpcErr) Data() interface{} {
	return e.Dat
}

// NewJsonRpcErr creates new JSON-RPC error.
//
// code is the JSON-RPC error code.
// message is optional in case of well known error code (negative values).
// data is an optional extra info object.
func NewJsonRpcErr(code int, message string, data interface{}) error {
	if code < 0 {
		if message != "" {
			message = fmt.Sprintf("%s:[%s]", ErrStr[code], message)
		} else {
			message = ErrStr[code]
		}
	}

	return &JsonRpcErr{Cod: code, Mess: message, Dat: data}
}

type JsonRpcReq struct {
	Jsonrpc string      `json:"jsonrpc"`
	Id      uint64      `json:"id,omitempty"`
	Method  string      `json:"method"`
	Params  interface{} `json:"params"`
	nc      *NexusConn
}

type JsonRpcRes struct {
	Jsonrpc string      `json:"jsonrpc"`
	Id      uint64      `json:"id,omitempty"`
	Result  interface{} `json:"result,omitempty"`
	Error   *JsonRpcErr `json:"error,omitempty"`
}

// NexusConn represents the Nexus connection.
type NexusConn struct {
	conn         net.Conn
	connRx       *smartio.SmartReader
	connTx       *smartio.SmartWriter
	reqCount     uint64
	resTable     map[uint64]chan *JsonRpcRes
	resTableLock sync.Mutex
	chReq        chan *JsonRpcReq
	closed       int32 //Goroutine safe bool
	context      context.Context
	cancelFun    context.CancelFunc
	wdog         int64
}

// Task represents a task pushed to Nexus.
type Task struct {
	nc     *NexusConn
	taskId string
	Path   string
	Method string
	Params interface{}
	Prio   int
	Detach bool
	User   string
	Tags   map[string]interface{}
}

// TaskOpts represents task push options.
type TaskOpts struct {
	// Task priority default 0 (Set negative value for lower priority)
	Priority int
	// Task ttl default 5
	Ttl int
	// Task detach. If true, task is detached from creating session.
	// If task is detached and creating session deads, task is not removed from tasks queue.
	Detach bool
}

// Pipe represents a pipe.
// Pipes can only be read from the connection that created them and
// can be written from any conection.
type Pipe struct {
	nc        *NexusConn
	pipeId    string
	context   context.Context
	cancelFun context.CancelFunc
}

// Msg represents a pipe single message
type Msg struct {
	Count int64       // Message counter (unique and correlative)
	Msg   interface{} // Pipe message
}

// PipeData represents a pipe messages group obtained in read ops.
type PipeData struct {
	Msgs    []*Msg // Messages
	Waiting int    // Number of messages waiting in Nexus server since last read
	Drops   int    // Number of messages dropped (pipe overflows) since last read
}

// PipeOpts represents pipe creation options
type PipeOpts struct {
	Length int // Pipe buffer capacity
}

// NewNexusConn creates new nexus connection from net.conn
func NewNexusConn(conn net.Conn) *NexusConn {
	nc := &NexusConn{
		conn:     conn,
		connRx:   smartio.NewSmartReader(conn),
		connTx:   smartio.NewSmartWriter(conn),
		resTable: make(map[uint64]chan *JsonRpcRes),
		chReq:    make(chan *JsonRpcReq, 16),
		wdog:     60,
	}
	nc.context, nc.cancelFun = context.WithCancel(context.Background())
	go nc.sendWorker()
	go nc.recvWorker()
	go nc.mainWorker()
	return nc
}

func (nc *NexusConn) pushReq(req *JsonRpcReq) (err error) {
	select {
	case nc.chReq <- req:
	case <-nc.context.Done():
		err = NewJsonRpcErr(ErrCancel, "", nil)
	}
	return
}

func (nc *NexusConn) pullReq() (req *JsonRpcReq, err error) {
	select {
	case req = <-nc.chReq:
	case <-nc.context.Done():
		err = NewJsonRpcErr(ErrCancel, "", nil)
	}
	return
}

func (nc *NexusConn) mainWorker() {
	defer nc.Close()
	tick := time.NewTicker(10 * time.Second)
	defer tick.Stop()
	for {
		select {
		case <-tick.C:
			now := time.Now().Unix()
			wdog := atomic.LoadInt64(&nc.wdog)
			max := wdog * 2
			if (now-nc.connRx.GetLast() > max) &&
				(now-nc.connTx.GetLast() > max) {
				return
			}
			if (now-nc.connRx.GetLast() >= wdog) &&
				(now-nc.connTx.GetLast() >= wdog) {
				if err := nc.Ping(10 * time.Second); err != nil {
					if time.Now().Unix()-nc.connTx.GetLast() > 5 {
						return
					}
				}
			}
		case <-nc.context.Done():
			return
		}
	}
}

func (nc *NexusConn) sendWorker() {
	defer nc.Close()
	for {
		req, err := nc.pullReq()
		if err != nil {
			break
		}
		req.Jsonrpc = "2.0"
		buf, err := json.Marshal(req)
		if err != nil {
			break
		}
		if !nc.Closed() {
			_, err = nc.connTx.Write(buf)
			if err != nil {
				break
			}
		} else {
			break
		}
	}
}

func (nc *NexusConn) recvWorker() {
	defer nc.Close()
	dec := json.NewDecoder(nc.connRx)
	for dec.More() {
		res := &JsonRpcRes{}
		err := dec.Decode(res)
		if err != nil {
			break
		}
		nc.resTableLock.Lock()
		ch := nc.resTable[res.Id]
		nc.resTableLock.Unlock()
		if ch != nil {
			select {
			case ch <- res:
			default:
			}
		}
	}
}

func (nc *NexusConn) newId() (uint64, chan *JsonRpcRes) {
	id := atomic.AddUint64(&nc.reqCount, 1)
	ch := make(chan *JsonRpcRes, 1)
	nc.resTableLock.Lock()
	defer nc.resTableLock.Unlock()
	nc.resTable[id] = ch
	return id, ch
}

func (nc *NexusConn) delId(id uint64) {
	nc.resTableLock.Lock()
	delete(nc.resTable, id)
	nc.resTableLock.Unlock()
}

// GetContext returns internal connection context.
func (nc *NexusConn) GetContext() context.Context {
	return nc.context
}

// Close closes nexus connection.
func (nc *NexusConn) Close() {
	if atomic.CompareAndSwapInt32(&nc.closed, 0, 1) {
		nc.cancelFun()
		nc.conn.Close()
	}
}

// Closed returns Nexus connection state.
func (nc *NexusConn) Closed() bool {
	return atomic.LoadInt32(&nc.closed) == 1
}

// ExecNoWait is a low level JSON-RPC call function, it don't wait response from server.
func (nc *NexusConn) ExecNoWait(method string, params interface{}) (id uint64, rch chan *JsonRpcRes, err error) {
	if nc.Closed() {
		err = NewJsonRpcErr(ErrCancel, "", nil)
		return
	}
	id, rch = nc.newId()
	req := &JsonRpcReq{
		Id:     id,
		Method: method,
		Params: params,
	}
	err = nc.pushReq(req)
	if err != nil {
		nc.delId(id)
		return 0, nil, err
	}
	return
}

// Exec is a low level JSON-RPC call function.
func (nc *NexusConn) Exec(method string, params interface{}) (result interface{}, err error) {
	id, rch, err := nc.ExecNoWait(method, params)
	if err != nil {
		return nil, err
	}
	defer nc.delId(id)
	select {
	case res := <-rch:
		if res.Error != nil {
			err = res.Error
		} else {
			result = res.Result
		}
	case <-nc.context.Done():
		err = NewJsonRpcErr(ErrCancel, "", nil)
	}
	return
}

// Ping pings Nexus server, timeout is the max time waiting for server response,
// after that ErrTimeout is returned.
func (nc *NexusConn) Ping(timeout time.Duration) (err error) {
	id, rch, err := nc.ExecNoWait("sys.ping", nil)
	if err != nil {
		return
	}
	defer nc.delId(id)
	select {
	case <-rch:
		break
	case <-time.After(timeout):
		err = NewJsonRpcErr(ErrTimeout, "", nil)
	case <-nc.context.Done():
		err = NewJsonRpcErr(ErrCancel, "", nil)
	}
	return
}

// Login attempts to login using user and pass.
// Returns the response object from Nexus or error.
func (nc *NexusConn) Login(user string, pass string) (interface{}, error) {
	par := map[string]interface{}{
		"user": user,
		"pass": pass,
	}
	return nc.Exec("sys.login", par)

}

type UserSessions struct {
	User     string        `json:"user"`
	Sessions []SessionInfo `json:"sessions"`
	N        int           `json:"n"`
}

type SessionInfo struct {
	Id            string    `json:"id"`
	NodeId        string    `json:"nodeId"`
	RemoteAddress string    `json:"remoteAddress"`
	Protocol      string    `json:"protocol"`
	CreationTime  time.Time `json:"creationTime"`
}

// Sessions returns info of the users sessions
// Returns a list of SessionInfo structs or an error
func (nc *NexusConn) SessionList(prefix string) ([]UserSessions, error) {
	par := map[string]interface{}{
		"prefix": prefix,
	}
	res, err := nc.Exec("sys.sessions.list", par)
	if err != nil {
		return nil, err
	}
	sessions := make([]UserSessions, 0)
	b, err := json.Marshal(res)
	if err != nil {
		return nil, err
	}
	err = json.Unmarshal(b, &sessions)
	if err != nil {
		return nil, err
	}

	return sessions, nil
}

// SessionKick forces the node owner of the client connection to close it
// Returns the response object from Nexus or error.
func (nc *NexusConn) SessionKick(connId string) (interface{}, error) {
	par := map[string]interface{}{
		"connId": connId,
	}
	return nc.Exec("sys.sessions.kick", par)
}

type NodeInfo struct {
	Load    map[string]float64 `json:"load"`
	Clients int                `json:"clients"`
	NodeId  string             `json:"id"`
}

// Nodes returns info of the nodes state
// Returns a list of NodeInfo structs or an error
func (nc *NexusConn) NodeList() ([]NodeInfo, error) {
	par := map[string]interface{}{}
	res, err := nc.Exec("sys.nodes.list", par)
	if err != nil {
		return nil, err
	}
	nodes := make([]NodeInfo, 0)
	b, err := json.Marshal(res)
	if err != nil {
		return nil, err
	}
	err = json.Unmarshal(b, &nodes)<commit_msg>Rename methods to singular<commit_after>package nxcore

import (
	"encoding/json"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"

	"github.com/jaracil/ei"
	"github.com/jaracil/smartio"
	"golang.org/x/net/context"
)

const (
	ErrParse            = -32700
	ErrInvalidRequest   = -32600
	ErrMethodNotFound   = -32601
	ErrInvalidParams    = -32602
	ErrInternal         = -32603
	ErrTimeout          = -32000
	ErrCancel           = -32001
	ErrInvalidTask      = -32002
	ErrInvalidPipe      = -32003
	ErrInvalidUser      = -32004
	ErrUserExists       = -32005
	ErrPermissionDenied = -32010
	ErrTtlExpired       = -32011
)

var ErrStr = map[int]string{
	ErrParse:            "Parse error",
	ErrInvalidRequest:   "Invalid request",
	ErrMethodNotFound:   "Method not found",
	ErrInvalidParams:    "Invalid params",
	ErrInternal:         "Internal error",
	ErrTimeout:          "Timeout",
	ErrCancel:           "Cancel",
	ErrInvalidTask:      "Invalid task",
	ErrInvalidPipe:      "Invalid pipe",
	ErrInvalidUser:      "Invalid user",
	ErrUserExists:       "User already exists",
	ErrPermissionDenied: "Permission denied",
	ErrTtlExpired:       "TTL expired",
}

type JsonRpcErr struct {
	Cod  int         `json:"code"`
	Mess string      `json:"message"`
	Dat  interface{} `json:"data,omitempty"`
}

func (e *JsonRpcErr) Error() string {
	return fmt.Sprintf("[%d] %s", e.Cod, e.Mess)
}

func (e *JsonRpcErr) Code() int {
	return e.Cod
}

func (e *JsonRpcErr) Data() interface{} {
	return e.Dat
}

// NewJsonRpcErr creates new JSON-RPC error.
//
// code is the JSON-RPC error code.
// message is optional in case of well known error code (negative values).
// data is an optional extra info object.
func NewJsonRpcErr(code int, message string, data interface{}) error {
	if code < 0 {
		if message != "" {
			message = fmt.Sprintf("%s:[%s]", ErrStr[code], message)
		} else {
			message = ErrStr[code]
		}
	}

	return &JsonRpcErr{Cod: code, Mess: message, Dat: data}
}

type JsonRpcReq struct {
	Jsonrpc string      `json:"jsonrpc"`
	Id      uint64      `json:"id,omitempty"`
	Method  string      `json:"method"`
	Params  interface{} `json:"params"`
	nc      *NexusConn
}

type JsonRpcRes struct {
	Jsonrpc string      `json:"jsonrpc"`
	Id      uint64      `json:"id,omitempty"`
	Result  interface{} `json:"result,omitempty"`
	Error   *JsonRpcErr `json:"error,omitempty"`
}

// NexusConn represents the Nexus connection.
type NexusConn struct {
	conn         net.Conn
	connRx       *smartio.SmartReader
	connTx       *smartio.SmartWriter
	reqCount     uint64
	resTable     map[uint64]chan *JsonRpcRes
	resTableLock sync.Mutex
	chReq        chan *JsonRpcReq
	closed       int32 //Goroutine safe bool
	context      context.Context
	cancelFun    context.CancelFunc
	wdog         int64
}

// Task represents a task pushed to Nexus.
type Task struct {
	nc     *NexusConn
	taskId string
	Path   string
	Method string
	Params interface{}
	Prio   int
	Detach bool
	User   string
	Tags   map[string]interface{}
}

// TaskOpts represents task push options.
type TaskOpts struct {
	// Task priority default 0 (Set negative value for lower priority)
	Priority int
	// Task ttl default 5
	Ttl int
	// Task detach. If true, task is detached from creating session.
	// If task is detached and creating session deads, task is not removed from tasks queue.
	Detach bool
}

// Pipe represents a pipe.
// Pipes can only be read from the connection that created them and
// can be written from any conection.
type Pipe struct {
	nc        *NexusConn
	pipeId    string
	context   context.Context
	cancelFun context.CancelFunc
}

// Msg represents a pipe single message
type Msg struct {
	Count int64       // Message counter (unique and correlative)
	Msg   interface{} // Pipe message
}

// PipeData represents a pipe messages group obtained in read ops.
type PipeData struct {
	Msgs    []*Msg // Messages
	Waiting int    // Number of messages waiting in Nexus server since last read
	Drops   int    // Number of messages dropped (pipe overflows) since last read
}

// PipeOpts represents pipe creation options
type PipeOpts struct {
	Length int // Pipe buffer capacity
}

// NewNexusConn creates new nexus connection from net.conn
func NewNexusConn(conn net.Conn) *NexusConn {
	nc := &NexusConn{
		conn:     conn,
		connRx:   smartio.NewSmartReader(conn),
		connTx:   smartio.NewSmartWriter(conn),
		resTable: make(map[uint64]chan *JsonRpcRes),
		chReq:    make(chan *JsonRpcReq, 16),
		wdog:     60,
	}
	nc.context, nc.cancelFun = context.WithCancel(context.Background())
	go nc.sendWorker()
	go nc.recvWorker()
	go nc.mainWorker()
	return nc
}

func (nc *NexusConn) pushReq(req *JsonRpcReq) (err error) {
	select {
	case nc.chReq <- req:
	case <-nc.context.Done():
		err = NewJsonRpcErr(ErrCancel, "", nil)
	}
	return
}

func (nc *NexusConn) pullReq() (req *JsonRpcReq, err error) {
	select {
	case req = <-nc.chReq:
	case <-nc.context.Done():
		err = NewJsonRpcErr(ErrCancel, "", nil)
	}
	return
}

func (nc *NexusConn) mainWorker() {
	defer nc.Close()
	tick := time.NewTicker(10 * time.Second)
	defer tick.Stop()
	for {
		select {
		case <-tick.C:
			now := time.Now().Unix()
			wdog := atomic.LoadInt64(&nc.wdog)
			max := wdog * 2
			if (now-nc.connRx.GetLast() > max) &&
				(now-nc.connTx.GetLast() > max) {
				return
			}
			if (now-nc.connRx.GetLast() >= wdog) &&
				(now-nc.connTx.GetLast() >= wdog) {
				if err := nc.Ping(10 * time.Second); err != nil {
					if time.Now().Unix()-nc.connTx.GetLast() > 5 {
						return
					}
				}
			}
		case <-nc.context.Done():
			return
		}
	}
}

func (nc *NexusConn) sendWorker() {
	defer nc.Close()
	for {
		req, err := nc.pullReq()
		if err != nil {
			break
		}
		req.Jsonrpc = "2.0"
		buf, err := json.Marshal(req)
		if err != nil {
			break
		}
		if !nc.Closed() {
			_, err = nc.connTx.Write(buf)
			if err != nil {
				break
			}
		} else {
			break
		}
	}
}

func (nc *NexusConn) recvWorker() {
	defer nc.Close()
	dec := json.NewDecoder(nc.connRx)
	for dec.More() {
		res := &JsonRpcRes{}
		err := dec.Decode(res)
		if err != nil {
			break
		}
		nc.resTableLock.Lock()
		ch := nc.resTable[res.Id]
		nc.resTableLock.Unlock()
		if ch != nil {
			select {
			case ch <- res:
			default:
			}
		}
	}
}

func (nc *NexusConn) newId() (uint64, chan *JsonRpcRes) {
	id := atomic.AddUint64(&nc.reqCount, 1)
	ch := make(chan *JsonRpcRes, 1)
	nc.resTableLock.Lock()
	defer nc.resTableLock.Unlock()
	nc.resTable[id] = ch
	return id, ch
}

func (nc *NexusConn) delId(id uint64) {
	nc.resTableLock.Lock()
	delete(nc.resTable, id)
	nc.resTableLock.Unlock()
}

// GetContext returns internal connection context.
func (nc *NexusConn) GetContext() context.Context {
	return nc.context
}

// Close closes nexus connection.
func (nc *NexusConn) Close() {
	if atomic.CompareAndSwapInt32(&nc.closed, 0, 1) {
		nc.cancelFun()
		nc.conn.Close()
	}
}

// Closed returns Nexus connection state.
func (nc *NexusConn) Closed() bool {
	return atomic.LoadInt32(&nc.closed) == 1
}

// ExecNoWait is a low level JSON-RPC call function, it don't wait response from server.
func (nc *NexusConn) ExecNoWait(method string, params interface{}) (id uint64, rch chan *JsonRpcRes, err error) {
	if nc.Closed() {
		err = NewJsonRpcErr(ErrCancel, "", nil)
		return
	}
	id, rch = nc.newId()
	req := &JsonRpcReq{
		Id:     id,
		Method: method,
		Params: params,
	}
	err = nc.pushReq(req)
	if err != nil {
		nc.delId(id)
		return 0, nil, err
	}
	return
}

// Exec is a low level JSON-RPC call function.
func (nc *NexusConn) Exec(method string, params interface{}) (result interface{}, err error) {
	id, rch, err := nc.ExecNoWait(method, params)
	if err != nil {
		return nil, err
	}
	defer nc.delId(id)
	select {
	case res := <-rch:
		if res.Error != nil {
			err = res.Error
		} else {
			result = res.Result
		}
	case <-nc.context.Done():
		err = NewJsonRpcErr(ErrCancel, "", nil)
	}
	return
}

// Ping pings Nexus server, timeout is the max time waiting for server response,
// after that ErrTimeout is returned.
func (nc *NexusConn) Ping(timeout time.Duration) (err error) {
	id, rch, err := nc.ExecNoWait("sys.ping", nil)
	if err != nil {
		return
	}
	defer nc.delId(id)
	select {
	case <-rch:
		break
	case <-time.After(timeout):
		err = NewJsonRpcErr(ErrTimeout, "", nil)
	case <-nc.context.Done():
		err = NewJsonRpcErr(ErrCancel, "", nil)
	}
	return
}

// Login attempts to login using user and pass.
// Returns the response object from Nexus or error.
func (nc *NexusConn) Login(user string, pass string) (interface{}, error) {
	par := map[string]interface{}{
		"user": user,
		"pass": pass,
	}
	return nc.Exec("sys.login", par)

}

type UserSessions struct {
	User     string        `json:"user"`
	Sessions []SessionInfo `json:"sessions"`
	N        int           `json:"n"`
}

type SessionInfo struct {
	Id            string    `json:"id"`
	NodeId        string    `json:"nodeId"`
	RemoteAddress string    `json:"remoteAddress"`
	Protocol      string    `json:"protocol"`
	CreationTime  time.Time `json:"creationTime"`
}

// Sessions returns info of the users sessions
// Returns a list of SessionInfo structs or an error
func (nc *NexusConn) SessionList(prefix string) ([]UserSessions, error) {
	par := map[string]interface{}{
		"prefix": prefix,
	}
	res, err := nc.Exec("sys.sessions.list", par)
	res, err := nc.Exec("sys.session.list", par)
	if err != nil {
		return nil, err
	}
	sessions := make([]UserSessions, 0)
	b, err := json.Marshal(res)
	if err != nil {
		return nil, err
	}
	err = json.Unmarshal(b, &sessions)
	if err != nil {
		return nil, err
	}

	return sessions, nil
}

// SessionKick forces the node owner of the client connection to close it
// Returns the response object from Nexus or error.
func (nc *NexusConn) SessionKick(connId string) (interface{}, error) {
	par := map[string]interface{}{
		"connId": connId,
	}
	return nc.Exec("sys.sessions.kick", par)
	return nc.Exec("sys.session.kick", par)
}

type NodeInfo struct {
	Load    map[string]float64 `json:"load"`
	Clients int                `json:"clients"`
	NodeId  string             `json:"id"`
}

// Nodes returns info of the nodes state
// Returns a list of NodeInfo structs or an error
func (nc *NexusConn) NodeList() ([]NodeInfo, error) {
	par := map[string]interface{}{}
	res, err := nc.Exec("sys.nodes.list", par)
	res, err := nc.Exec("sys.node.list", par)
	if err != nil {
		return nil, err
	}
	nodes := make([]NodeInfo, 0)
	b, err := json.Marshal(res)
	if err != nil {
		return nil, err
	}
	err = json.Unmarshal(b, &nodes)
{'fraction_non_alphanumeric': 0.12985768364559874, 'fraction_numerical': 0.011500311466768892, 'mean_word_length': 3.1269527387779315, 'pattern_counts': {'":': 8, '<': 29, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5348242', 'n_tokens_mistral': 7944, 'n_tokens_neox': 7399, 'n_words': 2127}